stan

Apr 6, 2021

Changes in this version

+ func MetadataContextDecorator() func(context.Context, binding.Message) context.Context

+ type MsgMetadata struct

+ Redelivered bool

+ RedeliveryCount uint32

+ Sequence uint64

+ func MessageMetadataFrom(ctx context.Context) (MsgMetadata, bool)

May 27, 2020

Changes in this version

+ var ErrInvalidQueueName = errors.New("invalid queue name for QueueSubscriber")

+ func StanOptions(opts ...stan.Option) []stan.Option

+ func WriteMsg(ctx context.Context, m binding.Message, writer io.ReaderFrom, ...) error

+ type Consumer struct

+ Conn stan.Conn

+ Subject string

+ Subscriber Subscriber

+ UnsubscribeOnClose bool

+ func NewConsumer(clusterID, clientID, subject string, stanOpts []stan.Option, ...) (*Consumer, error)

+ func NewConsumerFromConn(conn stan.Conn, subject string, opts ...ConsumerOption) (*Consumer, error)

+ func (c *Consumer) Close(_ context.Context) error

+ func (c *Consumer) OpenInbound(ctx context.Context) error

+ type ConsumerOption func(*Consumer) error

+ func WithQueueSubscriber(queue string) ConsumerOption

+ func WithSubscriptionOptions(opts ...stan.SubscriptionOption) ConsumerOption

+ func WithUnsubscribeOnClose() ConsumerOption

+ type Message struct

+ Msg *stan.Msg

+ func NewMessage(msg *stan.Msg, opts ...MessageOption) (*Message, error)

+ func (m *Message) Finish(err error) error

+ func (m *Message) ReadBinary(context.Context, binding.BinaryWriter) error

+ func (m *Message) ReadEncoding() binding.Encoding

+ func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error

+ type MessageOption func(*Message) error

+ func WithManualAcks() MessageOption

+ type Protocol struct

+ Conn stan.Conn

+ Consumer *Consumer

+ Sender *Sender

+ func NewProtocol(clusterID, clientID, sendSubject, receiveSubject string, ...) (*Protocol, error)

+ func NewProtocolFromConn(conn stan.Conn, sendSubject, receiveSubject string, opts ...ProtocolOption) (*Protocol, error)

+ func (p *Protocol) Close(ctx context.Context) (err error)

+ func (p *Protocol) OpenInbound(ctx context.Context) error

+ func (p *Protocol) Receive(ctx context.Context) (binding.Message, error)

+ func (p *Protocol) Send(ctx context.Context, in binding.Message, transformers ...binding.Transformer) error

+ type ProtocolOption func(*Protocol) error

+ func WithConsumerOptions(opts ...ConsumerOption) ProtocolOption

+ func WithSenderOptions(opts ...SenderOption) ProtocolOption

+ type QueueSubscriber struct

+ QueueGroup string

+ func (s *QueueSubscriber) Subscribe(conn stan.Conn, subject string, cb stan.MsgHandler, ...) (stan.Subscription, error)

+ type Receiver struct

+ func NewReceiver(opts ...ReceiverOption) (*Receiver, error)

+ func (r *Receiver) MsgHandler(msg *stan.Msg)

+ func (r *Receiver) Receive(ctx context.Context) (binding.Message, error)

+ type ReceiverOption func(*Receiver) error

+ func WithMessageOptions(opts ...MessageOption) ReceiverOption

+ type RegularSubscriber struct

+ func (s *RegularSubscriber) Subscribe(conn stan.Conn, subject string, cb stan.MsgHandler, ...) (stan.Subscription, error)

+ type Sender struct

+ Conn stan.Conn

+ Subject string

+ func NewSender(clusterID, clientID, subject string, stanOpts []stan.Option, ...) (*Sender, error)

+ func NewSenderFromConn(conn stan.Conn, subject string, opts ...SenderOption) (*Sender, error)

+ func (s *Sender) Close(_ context.Context) error

+ func (s *Sender) Send(ctx context.Context, in binding.Message, transformers ...binding.Transformer) (err error)

+ type SenderOption func(*Sender) error

+ type Subscriber interface

+ Subscribe func(conn stan.Conn, subject string, cb stan.MsgHandler, ...) (stan.Subscription, error)