jetstream

Documentation

Index

• type MarshalFn
• type Option
  • func WithConfig(v *nats.StreamConfig) Option
  • func WithNamespace(v string) Option
  • func WithNatsOptions(v ...nats.Option) Option
  • func WithReconnectFailedHandler(v func(error)) Option
  • func WithReconnectMaxRetries(v int) Option
  • func WithReconnectTimeout(v time.Duration) Option
• type Publisher
  • func (s *Publisher) JS() nats.JetStreamContext
  • func (s *Publisher) Marshal(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 PublisherWithMarshal(marshal MarshalFn) PublisherOption
  • func PublisherWithPubOpts(v ...nats.PubOpt) PublisherOption
• type Stream
  • func New(l *zap.Logger, name, addr string, opts ...Option) (*Stream, error)
  • func (s *Stream) Addr() string
  • func (s *Stream) Close()
  • 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) ChanSubscribe(ch chan *nats.Msg, opts ...nats.SubOpt) (*nats.Subscription, error)
  • 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) Unmarshal(msg *nats.Msg, v interface{}) error
• type SubscriberOption
  • func SubscriberWithNamespace(v string) SubscriberOption
  • func SubscriberWithSubOpts(v ...nats.SubOpt) SubscriberOption
  • func SubscriberWithUnmarshal(unmarshal UnmarshalFn) SubscriberOption
• type UnmarshalFn

Types

type MarshalFn

type MarshalFn func(v interface{}) ([]byte, 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

func WithReconnectFailedHandler

func WithReconnectFailedHandler(v func(error)) Option

WithReconnectFailedHandler option

func WithReconnectMaxRetries

func WithReconnectMaxRetries(v int) Option

WithReconnectMaxRetries option

func WithReconnectTimeout

func WithReconnectTimeout(v time.Duration) Option

WithReconnectTimeout option

type Publisher

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

func (*Publisher) JS

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

func (*Publisher) Marshal

func (s *Publisher) Marshal(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 PublisherWithMarshal

func PublisherWithMarshal(marshal MarshalFn) 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) Addr

func (s *Stream) Addr() string

func (*Stream) Close

func (s *Stream) Close()

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) ChanSubscribe

func (s *Subscriber) ChanSubscribe(ch chan *nats.Msg, opts ...nats.SubOpt) (*nats.Subscription, error)

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) Unmarshal

func (s *Subscriber) Unmarshal(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

func SubscriberWithUnmarshal

func SubscriberWithUnmarshal(unmarshal UnmarshalFn) SubscriberOption

type UnmarshalFn

type UnmarshalFn func(data []byte, v interface{}) error