nats

Documentation

Index

• func Connect(url string) (*nats.Conn, error)
• func NewJetStreamContext(nc *nats.Conn, streams []nats.StreamConfig) (nats.JetStreamContext, error)
• func NewStreams(streams ...nats.StreamConfig) []nats.StreamConfig
• type ConsumerOption
  • func WithBindStream(stream string) ConsumerOption
  • func WithCloseOnDone() ConsumerOption
  • func WithDurable(durable string) ConsumerOption
  • func WithMaxWait(d time.Duration) ConsumerOption
  • func WithMustHaveHandler(must bool) ConsumerOption
• type ConsumerOptions
• type EventsConsumer
  • func NewEventsConsumer(nc *nats.Conn, handlers pubsub.Handlers, logger *zerolog.Logger, ...) (*EventsConsumer, error)
  • func (c *EventsConsumer) Consume(ctx context.Context, topic string) error
• type EventsPublisher
  • func NewEventsPublisher(nc *nats.Conn) *EventsPublisher
  • func (p *EventsPublisher) Publish(ctx context.Context, events []pubsub.Event) error
• type PersistentStreamConsumer
  • func NewPersistentStreamConsumer(nc nats.JetStreamContext, handlers pubsub.Handlers, logger *zerolog.Logger, ...) (*PersistentStreamConsumer, error)
  • func (c *PersistentStreamConsumer) Consume(ctx context.Context, subject string) error
• type StreamPublisher
  • func NewStreamPublisher(nc nats.JetStreamContext, stream string) *StreamPublisher
  • func (p *StreamPublisher) Publish(ctx context.Context, events []pubsub.Event) error

Functions

func Connect

func Connect(url string) (*nats.Conn, error)

func NewJetStreamContext

func NewJetStreamContext(nc *nats.Conn, streams []nats.StreamConfig) (nats.JetStreamContext, error)

func NewStreams

func NewStreams(streams ...nats.StreamConfig) []nats.StreamConfig

Types

type ConsumerOption

type ConsumerOption func(*ConsumerOptions)

func WithBindStream

func WithBindStream(stream string) ConsumerOption

func WithCloseOnDone

func WithCloseOnDone() ConsumerOption

func WithDurable

func WithDurable(durable string) ConsumerOption

func WithMaxWait

func WithMaxWait(d time.Duration) ConsumerOption

WithMaxWait specifies the maximum amount of time the consumer will wait for a message to arrive before returning.

func WithMustHaveHandler

func WithMustHaveHandler(must bool) ConsumerOption

WithMustHaveHandler specifies whether the consumer must have a handler for every message it receives.

type ConsumerOptions

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

type EventsConsumer

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

func NewEventsConsumer

func NewEventsConsumer(nc *nats.Conn, handlers pubsub.Handlers, logger *zerolog.Logger, opts ...ConsumerOption) (*EventsConsumer, error)

func (*EventsConsumer) Consume

func (c *EventsConsumer) Consume(ctx context.Context, topic string) error

type EventsPublisher

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

func NewEventsPublisher

func NewEventsPublisher(nc *nats.Conn) *EventsPublisher

func (*EventsPublisher) Publish

func (p *EventsPublisher) Publish(ctx context.Context, events []pubsub.Event) error

type PersistentStreamConsumer

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

func NewPersistentStreamConsumer

func NewPersistentStreamConsumer(nc nats.JetStreamContext, handlers pubsub.Handlers, logger *zerolog.Logger, opts ...ConsumerOption) (*PersistentStreamConsumer, error)

func (*PersistentStreamConsumer) Consume

func (c *PersistentStreamConsumer) Consume(ctx context.Context, subject string) error

type StreamPublisher

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

func NewStreamPublisher

func NewStreamPublisher(nc nats.JetStreamContext, stream string) *StreamPublisher

func (*StreamPublisher) Publish

func (p *StreamPublisher) Publish(ctx context.Context, events []pubsub.Event) error