stream

Documentation

Index

• type LocalStream
  • func NewLocalStream() (*LocalStream, error)
  • func (ls *LocalStream) CreateStream(ctx context.Context, name string, subjects []string) error
  • func (ls *LocalStream) Process() error
  • func (ls *LocalStream) Publish(ctx context.Context, topic string, data []byte) error
  • func (ls *LocalStream) Subscribe(ctx context.Context, _ string, topic string, handler func([]byte) error) error
• type NATSStream
  • func (ns *NATSStream) CreateStream(ctx context.Context, name string, subjects []string) error
  • func (ns *NATSStream) Publish(ctx context.Context, topic string, data []byte) error
  • func (ns *NATSStream) Subscribe(ctx context.Context, durable string, topic string, handler func([]byte) error) error
• type Stream
  • func NewNATSStream(config *config.Config) (Stream, error)

Types

type LocalStream

type LocalStream struct {
	Streams        map[string]func([]byte) error
	CreatedStreams []string
	Messages       map[string][][]byte
	FailToPublish  bool
	// contains filtered or unexported fields
}

func NewLocalStream

func NewLocalStream() (*LocalStream, error)

func (*LocalStream) CreateStream

func (ls *LocalStream) CreateStream(ctx context.Context, name string, subjects []string) error

func (*LocalStream) Process

func (ls *LocalStream) Process() error

func (*LocalStream) Publish

func (ls *LocalStream) Publish(ctx context.Context, topic string, data []byte) error

func (*LocalStream) Subscribe

func (ls *LocalStream) Subscribe(ctx context.Context, _ string, topic string, handler func([]byte) error) error

type NATSStream

type NATSStream struct {
	// contains filtered or unexported fields
}

func (*NATSStream) CreateStream

func (ns *NATSStream) CreateStream(ctx context.Context, name string, subjects []string) error

func (*NATSStream) Publish

func (ns *NATSStream) Publish(ctx context.Context, topic string, data []byte) error

func (*NATSStream) Subscribe

func (ns *NATSStream) Subscribe(ctx context.Context, durable string, topic string, handler func([]byte) error) error

type Stream

type Stream interface {
	CreateStream(ctx context.Context, name string, subjects []string) error
	Subscribe(ctx context.Context, durable string, topic string, handler func([]byte) error) error
	Publish(ctx context.Context, topic string, data []byte) error
}

func NewNATSStream

func NewNATSStream(config *config.Config) (Stream, error)