nats

Documentation

Index

• func New(url string, options ...func(*nats.Conn)) (nats.JetStreamContext, *nats.Conn, error)
• type Publisher
  • func NewPublisher(nc *nats.Conn, js nats.JetStreamContext) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, topic string, msg pubsub.Message) error
• type Subscriber
  • func NewSubscriber(queueGroup string, natsClient *nats.Conn, jetStreamCtx nats.JetStreamContext, ...) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Subscribe(ctx context.Context, subscription string, handler pubsub.Handler) error
  • func (s *Subscriber) SubscribeWithAck(ctx context.Context, subscription string, handler pubsub.HandlerWithAck) error

Functions

func New

func New(url string, options ...func(*nats.Conn)) (nats.JetStreamContext, *nats.Conn, error)

New returns JetStream context, nats connection and an error. Since we are most likely to only be using JetStream, there is no need to separate the initialization of NatsConn and JetStream

Types

type Publisher

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

Publisher publishes a message on a NATS JetStream Stream's Pub/Sub topic.

Subjects (topics) are managed by the server automatically following presence/absence of subscriptions https://docs.nats.io/reference/faq#how-do-i-create-subjects

For more info on how NATS JetStream work, check https://docs.nats.io/using-nats/developer/develop_jetstream.

func NewPublisher

func NewPublisher(nc *nats.Conn, js nats.JetStreamContext) (*Publisher, error)

NewPublisher create a new Nats JetStream publisher.

It required a call to Close in order to stop processing messages and close topic connections.

func (*Publisher) Close

func (p *Publisher) Close() error

Close notifies the Publisher to stop processing messages, send all the remaining messages and close the connection.

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, topic string, msg pubsub.Message) error

Publish publishes a message on a NATS Pub/Sub subject (topic).

It will be received by subscriber(s) in all cases, however to enable persistence of the message a Stream must be created JetStream publish calls are acknowledged by the JetStream enabled servers To receive messages published to a topic, you must create a subscription to that topic.

See https://docs.nats.io/nats-concepts/jetstream/streams to find out more about how NATS streams work.

type Subscriber

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

Subscriber is our wrapper around NATS subscription. In current implementation, one Subscriber corresponds to one NATS subscription, as it's ok to have many subscriptions per client(https://docs.nats.io/using-nats/developer/anatomy#connecting-and-disconnecting)

The following features are available our of the box: - automatic reconnection: https://docs.nats.io/using-nats/developer/connecting/reconnect

func NewSubscriber

func NewSubscriber(queueGroup string, natsClient *nats.Conn, jetStreamCtx nats.JetStreamContext, consumer *nats.ConsumerInfo) (*Subscriber, error)

NewSubscriber creates a new Nats Subscriber.

it required a call to Close in order to stop processing messages and close subscriber connections.

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close notifies the Subscriber to stop processing messages on all subscriptions, and terminate the connection.

It is caller's responsibility to configure client's connection's `DrainTimeout` and `ClosedHandler` (with WaitGroup) https://docs.nats.io/using-nats/developer/receiving/drain

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, subscription string, handler pubsub.Handler) error

Subscribe consumes NATS Pub/Sub.

NATS has two types of subscription: Pull and Push.

Read more about it https://docs.nats.io/reference/faq#what-is-the-right-kind-of-stream-consumer-to-use

IMPORTANT! Don't forget to filter messages on the consumer as subscriber's subscription doesn't seem to take priority. Depending on the Consumer `DeliverPolicy`, `all`, `last`, `new`, `by_start_time`, `by_start_sequence` persisted messages can be received

func (*Subscriber) SubscribeWithAck

func (s *Subscriber) SubscribeWithAck(ctx context.Context, subscription string, handler pubsub.HandlerWithAck) error