nats

type CommonConsumer ¶ added in v1.11.0

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

CommonConsumer is a base consumer for processing NATS messages.

func NewConsumer(
	url string,
	subject string,
	opts ...ConsumerOption,
) (*CommonConsumer, error)

func (c *CommonConsumer) Run() error

func (c *CommonConsumer) Stop() error

type CommonPublisher ¶

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

CommonPublisher is a base NATS publisher.

func NewPublisher(url string, opts ...natsbroker.Option) (*CommonPublisher, error)

func (p *CommonPublisher) Close() error

func (p *CommonPublisher) Publish(topic string, data []byte) error

type Consumer ¶ added in v1.11.0

type Consumer interface {
	Run() error
	Stop() error
}

Consumer asynchronously processes NATS messages in goroutines.

type ConsumerAlreadyRunningError ¶ added in v1.11.0

type ConsumerAlreadyRunningError struct {
	Message string
	BaseErr error
}

ConsumerAlreadyRunningError is an error, which represents, that consumer was already started and can not be started again.

func (e ConsumerAlreadyRunningError) Error() string

func (e ConsumerAlreadyRunningError) Unwrap() error

type ConsumerAlreadyStoppedError ¶ added in v1.11.0

type ConsumerAlreadyStoppedError struct {
	Message string
	BaseErr error
}

ConsumerAlreadyStoppedError is an error, which represents, that consumer was not started yet or was already stopped.

func (e ConsumerAlreadyStoppedError) Error() string

func (e ConsumerAlreadyStoppedError) Unwrap() error

type ConsumerOption ¶ added in v1.11.0

type ConsumerOption func(options *consumerOptions) error

ConsumerOption represents golang functional option pattern func for Consumer configuration.

func WithCloseHandler(handler func(connection *natsbroker.Conn)) ConsumerOption

func WithDisconnectErrorHandler(
	handler func(connection *natsbroker.Conn, err error),
) ConsumerOption

func WithErrorHandler(
	handler func(connection *natsbroker.Conn, subscription *natsbroker.Subscription, err error),
) ConsumerOption

func WithGoroutinesPoolSize(size int) ConsumerOption

func WithMessageChannelBufferSize(size int) ConsumerOption

func WithMessageHandler(handler func(message *natsbroker.Msg)) ConsumerOption

func WithNatsOptions(opts ...natsbroker.Option) ConsumerOption

type IDPublisher ¶ added in v1.17.0

type IDPublisher interface {
	Publisher

	// PublishWithID sends a message with a deduplication id. A repeated publish
	// with the same id inside the server deduplication window is dropped.
	PublishWithID(subject, msgID string, content []byte) error
}

IDPublisher publishes messages with a deduplication id.

type JetStreamConsumer ¶ added in v1.17.0

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

JetStreamConsumer processes messages of a durable JetStream consumer.

Reading position lives on the server, so a restarted worker continues from the place, where it stopped, and not from the current moment. Acknowledgement is manual: the handler acks after a successful delivery, and an unacked message is redelivered after AckWait.

func NewJetStreamConsumer(
	url string,
	subject string,
	opts ...JetStreamConsumerOption,
) (*JetStreamConsumer, error)

func (c *JetStreamConsumer) Run() error

func (c *JetStreamConsumer) Stop() error

type JetStreamConsumerOption ¶ added in v1.17.0

type JetStreamConsumerOption func(options *jetStreamConsumerOptions) error

JetStreamConsumerOption represents functional option for JetStreamConsumer.

func WithJetStreamAckWait(ackWait time.Duration) JetStreamConsumerOption

func WithJetStreamDurable(name string) JetStreamConsumerOption

func WithJetStreamGoroutinesPoolSize(size int) JetStreamConsumerOption

func WithJetStreamMaxAckPending(maxAckPending int) JetStreamConsumerOption

func WithJetStreamMaxDeliver(maxDeliver int) JetStreamConsumerOption

func WithJetStreamMessageChannelBufferSize(size int) JetStreamConsumerOption

func WithJetStreamMessageHandler(handler func(message *natsbroker.Msg)) JetStreamConsumerOption

func WithJetStreamNatsOptions(opts ...natsbroker.Option) JetStreamConsumerOption

func WithJetStreamStream(name string) JetStreamConsumerOption

type JetStreamPublisher ¶ added in v1.17.0

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

JetStreamPublisher publishes messages to a JetStream stream.

Unlike CommonPublisher it waits for a server acknowledgement: a publish, which returned nil, is stored on disk.

func NewJetStreamPublisher(
	url string,
	stream StreamConfig,
	opts ...natsbroker.Option,
) (*JetStreamPublisher, error)

func (p *JetStreamPublisher) Close() error

func (p *JetStreamPublisher) Publish(subject string, data []byte) error

func (p *JetStreamPublisher) PublishWithID(subject, msgID string, data []byte) error

type Publisher ¶

type Publisher interface {
	Publish(subject string, content []byte) error
	Close() error
}

Publisher publishes messages to NATS broker.

type StreamConfig ¶ added in v1.17.0

type StreamConfig struct {
	// Name is a stream name. Stream is created on publisher start, if missing.
	Name string

	// Subjects is a list of subject patterns, captured by the stream.
	Subjects []string

	// MaxAge is an upper bound of a message lifetime in the stream.
	MaxAge time.Duration

	// MaxBytes is an upper bound of a stream size on disk. Reaching it fails
	// a publish (see DiscardNew above), and does not drop a stored task.
	MaxBytes int64

	// Duplicates is a deduplication window for Nats-Msg-Id header.
	Duplicates time.Duration
}

StreamConfig describes a JetStream stream with file storage.

Retention is always WorkQueuePolicy: a message leaves the stream when it is acknowledged, not when it expires. MaxAge and MaxBytes are the upper bounds, which keep an undeliverable task from growing the stream forever.

Discard is always DiscardNew, and it is not a detail. With the default DiscardOld a stream, which hit MaxBytes, silently drops the OLDEST unacknowledged messages — that is, loses the very tasks the stream exists to keep. DiscardNew fails the publish instead, loudly and at the publisher.

Mind the WorkQueuePolicy constraint on the consumer side: filter subjects of consumers of such a stream must not overlap. One consumer per channel (`notifications.*.email`, `.web_push`, `.bell`) is fine; a second one on a subset of those subjects is refused with `filtered consumer not unique on workqueue stream`.

type StreamNotConfiguredError ¶ added in v1.17.0

type StreamNotConfiguredError struct {
	Message string
	BaseErr error
}

StreamNotConfiguredError is an error, which represents, that JetStream entity was created without a stream configuration.

func (e StreamNotConfiguredError) Error() string

func (e StreamNotConfiguredError) Unwrap() error