messaging

Mar 19, 2026

Changes in this version

+ const DefaultStreamMaxBytes

type BrokerOption

+ func WithMaxAckPending(maxAckPending int) BrokerOption

+ func WithMaxBytes(maxBytes int64) BrokerOption

Sep 4, 2025

Changes in this version

type DirectMessaging

+ SendToOther func(topic string, data []byte) error

+ SendToOtherWithRetry func(topic string, data []byte, config RetryConfig) error

+ SendToSelf func(topic string, data []byte) error

+ type RetryConfig struct

+ Delay time.Duration

+ ExponentialBackoff bool

+ OnRetry func(n uint, err error)

+ RetryAttempt uint

Jul 13, 2025

Changes in this version

+ const DefaultAckWait

+ const DefaultBackoffDuration

+ const DefaultConsumerPrefix

+ const DefaultMaxDeliveryAttempts

+ const DefaultStreamMaxAge

+ var ErrConnectionClosed = errors.New("connection is closed")

+ var ErrConsumerCreation = errors.New("failed to create consumer")

+ var ErrInvalidConfiguration = errors.New("invalid configuration")

+ var ErrInvalidStreamName = errors.New("stream name cannot be empty")

+ var ErrInvalidSubjects = errors.New("subjects cannot be empty")

+ var ErrStreamCreation = errors.New("failed to create stream")

+ var ErrSubscriptionClosed = errors.New("subscription is closed")

+ type BrokerOption func(*brokerConfiguration)

+ func WithBackoffDurations(durations []time.Duration) BrokerOption

+ func WithDeliverPolicy(policy jetstream.DeliverPolicy) BrokerOption

+ func WithRetentionPolicy(policy nats.RetentionPolicy) BrokerOption

+ func WithStorageType(storage nats.StorageType) BrokerOption

+ func WithStreamDescription(description string) BrokerOption

+ type MessageBroker interface

+ Close func() error

+ CreateSubscription func(ctx context.Context, consumerName, subject string, ...) (MessageSubscription, error)

+ FetchMessages func(ctx context.Context, consumerName, subject string, batchSize int, ...) error

+ GetStreamInfo func(ctx context.Context) (*jetstream.StreamInfo, error)

+ PublishMessage func(ctx context.Context, subject string, data []byte) error

+ func NewJetStreamBroker(ctx context.Context, natsConn *nats.Conn, streamName string, subjects []string, ...) (MessageBroker, error)

+ type MessageSubscription interface

+ Unsubscribe func() error

May 2, 2025

Changes in this version

+ var ErrPermament = errors.New("Permanent messaging error")

+ type DirectMessaging interface

+ Listen func(target string, handler func(data []byte)) (Subscription, error)

+ Send func(target string, data []byte) error

+ func NewNatsDirectMessaging(natsConn *nats.Conn) DirectMessaging

+ type EnqueueOptions struct

+ IdempotententKey string

+ type MessageQueue interface

+ Close func()

+ Dequeue func(topic string, handler func(message []byte) error) error

+ Enqueue func(topic string, message []byte, options *EnqueueOptions) error

+ type NATsMessageQueueManager struct

+ func NewNATsMessageQueueManager(queueName string, subjectWildCards []string, nc *nats.Conn) *NATsMessageQueueManager

+ func (m *NATsMessageQueueManager) NewMessageQueue(consumerName string) MessageQueue

+ type PubSub interface

+ Publish func(topic string, message []byte) error

+ PublishWithReply func(ttopic, reply string, data []byte) error

+ Subscribe func(topic string, handler func(msg *nats.Msg)) (Subscription, error)

+ func NewNATSPubSub(natsConn *nats.Conn) PubSub

+ type StreamPubsub interface

+ Publish func(topic string, message []byte) error

+ Subscribe func(name string, topic string, handler func(msg jetstream.Msg)) (Subscription, error)

+ func NewJetStreamPubSub(natsConn *nats.Conn, streamName string, subjects []string, ...) (StreamPubsub, error)

+ type StreamPubsubOption func(*streamPubSubConfig)

+ func WithAckWait(ackWait time.Duration) StreamPubsubOption

+ func WithConsumerNamePrefix(prefix string) StreamPubsubOption

+ func WithDescription(description string) StreamPubsubOption

+ func WithDiscardPolicy(policy nats.DiscardPolicy) StreamPubsubOption

+ func WithMaxAge(maxAge time.Duration) StreamPubsubOption

+ func WithMaxDeliveryAttempts(maxAttempts int) StreamPubsubOption

+ func WithRetention(policy nats.RetentionPolicy) StreamPubsubOption

+ func WithStorage(storage nats.StorageType) StreamPubsubOption

+ type Subscription interface

+ Unsubscribe func() error