jetstream

Documentation

Index

• type Initializer
• type Option
  • func WithConfig(v *nats.StreamConfig) Option
  • func WithNamespace(v string) Option
  • func WithNatsOptions(v ...nats.Option) Option
• type Publisher
  • func (s *Publisher) JS() nats.JetStreamContext
  • func (s *Publisher) Marshall(v interface{}) ([]byte, error)
  • func (s *Publisher) NewMsg(v interface{}) (*nats.Msg, error)
  • func (s *Publisher) PubOpts(opts ...nats.PubOpt) []nats.PubOpt
  • func (s *Publisher) PublishMsg(data interface{}, opts ...nats.PubOpt) (*nats.PubAck, error)
  • func (s *Publisher) PublishMsgAsync(data interface{}, opts ...nats.PubOpt) (nats.PubAckFuture, error)
  • func (s *Publisher) Subject() string
• type PublisherOption
  • func PublisherWithHeader(v nats.Header) PublisherOption
  • func PublisherWithPubOpts(v ...nats.PubOpt) PublisherOption
• type Stream
  • func New(l *zap.Logger, name, addr string, opts ...Option) (*Stream, error)
  • func (s *Stream) Conn() *nats.Conn
  • func (s *Stream) Info() *nats.StreamInfo
  • func (s *Stream) JS() nats.JetStreamContext
  • func (s *Stream) Name() string
  • func (s *Stream) Publisher(subject string, opts ...PublisherOption) *Publisher
  • func (s *Stream) Subscriber(subject string, opts ...SubscriberOption) *Subscriber
• type Subscriber
  • func (s *Subscriber) JS() nats.JetStreamContext
  • func (s *Subscriber) QueueSubscribe(queue string, handler stream.MsgHandler, opts ...nats.SubOpt) (*nats.Subscription, error)
  • func (s *Subscriber) SubOpts(opts ...nats.SubOpt) []nats.SubOpt
  • func (s *Subscriber) Subject() string
  • func (s *Subscriber) Subscribe(handler stream.MsgHandler, opts ...nats.SubOpt) (*nats.Subscription, error)
  • func (s *Subscriber) Unmarshall(msg *nats.Msg, v interface{}) error
• type SubscriberOption
  • func SubscriberWithNamespace(v string) SubscriberOption
  • func SubscriberWithSubOpts(v ...nats.SubOpt) SubscriberOption

Types

type Initializer

type Initializer func(*zap.Logger, nats.JetStreamContext) error

type Option

type Option func(*Stream)

func WithConfig

func WithConfig(v *nats.StreamConfig) Option

func WithNamespace

func WithNamespace(v string) Option

WithNamespace option

func WithNatsOptions

func WithNatsOptions(v ...nats.Option) Option

WithNatsOptions option

type Publisher

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

func (*Publisher) JS

func (s *Publisher) JS() nats.JetStreamContext

func (*Publisher) Marshall

func (s *Publisher) Marshall(v interface{}) ([]byte, error)

func (*Publisher) NewMsg

func (s *Publisher) NewMsg(v interface{}) (*nats.Msg, error)

func (*Publisher) PubOpts

func (s *Publisher) PubOpts(opts ...nats.PubOpt) []nats.PubOpt

func (*Publisher) PublishMsg

func (s *Publisher) PublishMsg(data interface{}, opts ...nats.PubOpt) (*nats.PubAck, error)

func (*Publisher) PublishMsgAsync

func (s *Publisher) PublishMsgAsync(data interface{}, opts ...nats.PubOpt) (nats.PubAckFuture, error)

func (*Publisher) Subject

func (s *Publisher) Subject() string

type PublisherOption

type PublisherOption func(*Publisher)

func PublisherWithHeader

func PublisherWithHeader(v nats.Header) PublisherOption

func PublisherWithPubOpts

func PublisherWithPubOpts(v ...nats.PubOpt) PublisherOption

type Stream

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

func New

func New(l *zap.Logger, name, addr string, opts ...Option) (*Stream, error)

func (*Stream) Conn

func (s *Stream) Conn() *nats.Conn

func (*Stream) Info

func (s *Stream) Info() *nats.StreamInfo

func (*Stream) JS

func (s *Stream) JS() nats.JetStreamContext

func (*Stream) Name

func (s *Stream) Name() string

func (*Stream) Publisher

func (s *Stream) Publisher(subject string, opts ...PublisherOption) *Publisher

func (*Stream) Subscriber

func (s *Stream) Subscriber(subject string, opts ...SubscriberOption) *Subscriber

type Subscriber

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

func (*Subscriber) JS

func (s *Subscriber) JS() nats.JetStreamContext

func (*Subscriber) QueueSubscribe

func (s *Subscriber) QueueSubscribe(queue string, handler stream.MsgHandler, opts ...nats.SubOpt) (*nats.Subscription, error)

func (*Subscriber) SubOpts

func (s *Subscriber) SubOpts(opts ...nats.SubOpt) []nats.SubOpt

func (*Subscriber) Subject

func (s *Subscriber) Subject() string

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(handler stream.MsgHandler, opts ...nats.SubOpt) (*nats.Subscription, error)

func (*Subscriber) Unmarshall

func (s *Subscriber) Unmarshall(msg *nats.Msg, v interface{}) error

type SubscriberOption

type SubscriberOption func(*Subscriber)

func SubscriberWithNamespace

func SubscriberWithNamespace(v string) SubscriberOption

func SubscriberWithSubOpts

func SubscriberWithSubOpts(v ...nats.SubOpt) SubscriberOption