jetstreamext

README

NATS JetStream Extensions

JetStream Extensions is a set of utilities providing additional features to jetstream package in nats.go client.

Installation

go get github.com/synadia-io/orbit.go/jetstreamext

Utilities

GetBatch and GetLastMsgsFor

GetBatch and GetLastMsgsFor are utilities that allow you to fetch multiple messages from a JetStream stream. Responses are returned in an iterator, which you can range over to receive messages.

GetBatch

GetBatch fetches a batch of messages from a provided stream, starting from either the lowest matching sequence, from the provided sequence, or from the given time. It can be configured to fetch messages from matching subject (which may contain wildcards) and up to a maximum byte limit.

Examples:

• fetching 10 messages from the beginning of the stream:

msgs, err := jetstreamext.GetBatch(ctx, js, "stream", 10)
if err != nil {
    // handle error
}
for msg, err := range msgs {
    if err != nil {
        // handle error
    }
    fmt.Println(string(msg.Data))
}

• fetching 10 messages from the stream starting from sequence 100 and matching subject:

msgs, err := jetstreamext.GetBatch(ctx, js, "stream", 10, jetstreamext.GetBatchSeq(100), jetstreamext.GetBatchSubject("foo"))
if err != nil {
    // handle error
}
// process msgs

• fetching 10 messages from the stream starting from time 1 hour ago:

msgs, err := jetstreamext.GetBatch(ctx, js, "stream", 10, jetstreamext.GetBatchStartTime(time.Now().Add(-time.Hour)))
if err != nil {
    // handle error
}
// process msgs

• fetching 10 messages or up to provided byte limit:

msgs, err := jetstreamext.GetBatch(ctx, js, "stream", 10, jetstreamext.GetBatchMaxBytes(1024))
if err != nil {
    // handle error
}
// process msgs

GetLastMsgsFor

GetLastMsgsFor fetches the last messages for the specified subjects from the specified stream. It can be optionally configured to fetch messages up to the provided sequence (or time), rather than the latest messages available. It can also be configured to fetch messages up to a provided batch size. The provided subjects may contain wildcards, however it is important to note that the NATS server will match a maximum of 1024 subjects.

Responses are returned in an iterator, which you can range over to receive messages.

Examples:

• fetching last messages from the stream for the provided subjects:

msgs, err := jetstreamext.GetLastMsgsFor(ctx, js, "stream", []string{"foo", "bar"})
if err != nil {
    // handle error
}
for msg, err := range msgs {
    if err != nil {
        // handle error
    }
    fmt.Println(string(msg.Data))
}

• fetching last messages from the stream for the provided subjects up to stream sequence 100:

msgs, err := jetstreamext.GetLastMsgsFor(ctx, js, "stream", []string{"foo", "bar"}, jetstreamext.GetLastMsgsUpToSeq(100))
if err != nil {
    // handle error
}
// process msgs

• fetching last messages from the stream for the provided subjects up to time 1 hour ago:

msgs, err := jetstreamext.GetLastMsgsFor(ctx, js, "stream", []string{"foo", "bar"}, jetstreamext.GetLastMsgsUpToTime(time.Now().Add(-time.Hour)))
if err != nil {
    // handle error
}
// process msgs

• fetching last messages from the stream for the provided subjects up to a batch size of 10:

msgs, err := jetstreamext.GetLastMsgsFor(ctx, js, "stream", []string{"foo.*"}, jetstreamext.GetLastMsgsBatchSize(10))
if err != nil {
    // handle error
}
// process msgs

Atomic batch publishing

PublishMsgBatch and BatchPublisher provide atomic batch publishing to JetStream streams with configurable flow control. A batch publish is an atomic operation - either all messages in the batch are persisted, or none are, depending on the result of the commit.

In order to use this feature, stream has to be configured with AllowAtomicPublish enabled.

Note: This module requires nats-server v2.12.0 o later.

BatchPublisher

BatchPublisher allows you to create a publisher that publishes messages in streaming-like fashion, where each message is published individually, but the commit is done for the entire batch. It can be configured with options for flow control and supports publish consistency checks. Adding messages to the batch is an IO operation, and messages are published immediately and persisted upon commit. A commit is done when the Commit method is called, which returns a BatchAck containing the results of the publish.

// Create a stream with batch publishing enabled

// stream has to be created with AllowAtomicPublish enabled
cfg := jetstream.StreamConfig{
    Name:               "FOO",
    Subjects:           []string{"foo.>"},
    AllowAtomicPublish: true,
}
stream, err := js.CreateStream(ctx, cfg)
if err != nil {
    // handle error
}

// Create a batch publisher
batch, err := jetstreamext.NewBatchPublisher(js)
if err != nil {
    // handle error
}

// Add message to the batch
err := batch.AddMsg("foo.A", &nats.Msg{
    Subject: "test.A",
    Data:    []byte("hello"),
})
if err != nil {
    // handle error
}

// Commit the batch
ack, err := batch.Commit(ctx, "test.A", []byte("commit msg"))
if err != nil {
    // handle error
}

By default, BatchPublisher waits for an ack from the server for the first message in the batch and for the commit. This can be configured with options, for example to wait for an ack for every 10 messages.

batch, err := jetstreamext.NewBatchPublisher(js, jetstreamext.BatchFlowControl{
    AckEvery:  10,
    AckTimeout: 5 * time.Second,
})
if err != nil {
    // handle error
}

PublishMsgBatch

PublishMsgBatch allows you to atomically publish a slice of messages to a stream and wait for an ack for the commit. It can be configured with options for flow control. For consistency checks, relevant headers can be set on individual messages.

msgs := make([]*nats.Msg, 0, count)
for range count {
    messages = append(messages, &nats.Msg{
        Subject: "foo",
        Data:    []byte("message"),
    })
}

ack, err := jetstreamext.PublishMsgBatch(ctx, js, messages)
if err != nil {
    // handle error
}

Documentation

Index

• Constants
• Variables
• func GetBatch(ctx context.Context, js jetstream.JetStream, stream string, batch int, ...) (iter.Seq2[*jetstream.RawStreamMsg, error], error)
• func GetLastMsgsFor(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, ...) (iter.Seq2[*jetstream.RawStreamMsg, error], error)
• type BatchAck
  • func PublishMsgBatch(ctx context.Context, js jetstream.JetStream, messages []*nats.Msg, ...) (*BatchAck, error)
• type BatchFlowControl
• type BatchMsgOpt
  • func WithBatchExpectLastSequence(seq uint64) BatchMsgOpt
  • func WithBatchExpectLastSequenceForSubject(seq uint64, subject string) BatchMsgOpt
  • func WithBatchExpectLastSequencePerSubject(seq uint64) BatchMsgOpt
  • func WithBatchExpectStream(stream string) BatchMsgOpt
  • func WithBatchMsgTTL(dur time.Duration) BatchMsgOpt
• type BatchPublisher
  • func NewBatchPublisher(js jetstream.JetStream, opts ...BatchPublisherOpt) (BatchPublisher, error)
• type BatchPublisherOpt
• type GetBatchOpt
  • func GetBatchMaxBytes(maxBytes int) GetBatchOpt
  • func GetBatchSeq(seq uint64) GetBatchOpt
  • func GetBatchStartTime(startTime time.Time) GetBatchOpt
  • func GetBatchSubject(subj string) GetBatchOpt
• type GetLastForOpt
  • func GetLastMsgsBatchSize(batch int) GetLastForOpt
  • func GetLastMsgsUpToSeq(seq uint64) GetLastForOpt
  • func GetLastMsgsUpToTime(tm time.Time) GetLastForOpt
• type PublishMsgBatchOpt

Constants

const (
	// Batch publish error codes
	JSErrCodeBatchPublishNotEnabled        jetstream.ErrorCode = 10174
	JSErrCodeBatchPublishMissingSeq        jetstream.ErrorCode = 10175
	JSErrCodeBatchPublishIncomplete        jetstream.ErrorCode = 10176
	JSErrCodeBatchPublishUnsupportedHeader jetstream.ErrorCode = 10177
	JSErrCodeBatchPublishExceedsLimit      jetstream.ErrorCode = 10199
)

const (
	// BatchIDHeader contains the batch ID for a message in a batch publish.
	BatchIDHeader = "Nats-Batch-Id"

	// BatchSeqHeader contains the sequence number of a message within a batch.
	BatchSeqHeader = "Nats-Batch-Sequence"

	// BatchCommitHeader signals the final message in a batch when set to "1".
	BatchCommitHeader = "Nats-Batch-Commit"
)

Variables

var (

	// ErrBatchPublishNotEnabled is returned when batch publish is not enabled on the stream.
	ErrBatchPublishNotEnabled jetstream.JetStreamError = &jsError{apiErr: &jetstream.APIError{ErrorCode: JSErrCodeBatchPublishNotEnabled, Description: "batch publish not enabled on stream", Code: 400}}

	// ErrBatchPublishIncomplete is returned when batch publish is incomplete and was abandoned.
	ErrBatchPublishIncomplete jetstream.JetStreamError = &jsError{apiErr: &jetstream.APIError{ErrorCode: JSErrCodeBatchPublishIncomplete, Description: "batch publish is incomplete and was abandoned", Code: 400}}

	// ErrBatchPublishMissingSeq is returned when batch publish sequence is missing.
	ErrBatchPublishMissingSeq jetstream.JetStreamError = &jsError{apiErr: &jetstream.APIError{ErrorCode: JSErrCodeBatchPublishMissingSeq, Description: "batch publish sequence is missing", Code: 400}}

	// ErrBatchPublishExceedsLimit is returned when batch publish sequence exceeds server limit (default 1000).
	ErrBatchPublishExceedsLimit jetstream.JetStreamError = &jsError{apiErr: &jetstream.APIError{ErrorCode: JSErrCodeBatchPublishExceedsLimit, Description: "batch publish sequence exceeds server limit (default 1000)", Code: 400}}

	// ErrBatchPublishUnsupportedHeader is returned when batch publish uses unsupported headers (Nats-Expected-Last-Msg-Id or Nats-Msg-Id).
	ErrBatchPublishUnsupportedHeader jetstream.JetStreamError = &jsError{apiErr: &jetstream.APIError{ErrorCode: JSErrCodeBatchPublishUnsupportedHeader, Description: "batch publish unsupported header used (Nats-Expected-Last-Msg-Id or Nats-Msg-Id)", Code: 400}}

	// ErrBatchClosed is returned when attempting to use a batch that has been closed.
	ErrBatchClosed = &jsError{message: "batch publisher closed"}

	// ErrInvalidBatchAck is returned when JetStream ack from batch publish is
	// invalid.
	ErrInvalidBatchAck jetstream.JetStreamError = &jsError{message: "invalid jetstream batch publish response"}
)

var (
	// ErrBatchUnsupported is returned when the server does not support batch
	// get (batch get is not supported by nats server >=2.11.0).
	ErrBatchUnsupported = errors.New("batch get not supported by server")

	// ErrInvalidResponse is returned when the response from the server is
	// invalid.
	ErrInvalidResponse = errors.New("invalid stream response")

	// ErrNoMessages is returned when there are no messages to fetch given the
	// provided options.
	ErrNoMessages = errors.New("no messages")

	// ErrInvalidOption is returned when an invalid option is provided.
	ErrInvalidOption = errors.New("invalid option")

	// ErrSubjectRequired is returned when no subjects are provided in GetLastMsgsFor.
	ErrSubjectRequired = errors.New("at least one subject is required")
)

Functions

func GetBatch

func GetBatch(ctx context.Context, js jetstream.JetStream, stream string, batch int, opts ...GetBatchOpt) (iter.Seq2[*jetstream.RawStreamMsg, error], error)

GetBatch fetches a batch of messages from the specified stream. The batch size is determined by the `batch` parameter. The function returns an iterator that can be used to iterate over the messages. Any error received during iteration will terminate the loop. The iterator will return an error if there are no messages to fetch.

func GetLastMsgsFor

func GetLastMsgsFor(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, opts ...GetLastForOpt) (iter.Seq2[*jetstream.RawStreamMsg, error], error)

GetLastMsgsFor fetches the last messages for the specified subjects from the specified stream. The function returns an iterator that can be used to iterate over the messages. Any error received during iteration will terminate the loop. It can be configured to fetch messages up to a certain stream sequence number or time.

Types

type BatchAck

type BatchAck struct {
	// Stream is the stream name the message was published to.
	Stream string `json:"stream"`

	// Sequence is the stream sequence number of the message.
	Sequence uint64 `json:"seq"`

	// Domain is the domain the message was published to.
	Domain string `json:"domain,omitempty"`

	// Value is the counter value for the stream.
	// This is only set when publishing to a stream with [StreamConfig.AllowMsgCounter] enabled.
	Value string `json:"val,omitempty"`

	// BatchID is the unique identifier for the batch.
	BatchID string `json:"batch_id,omitempty"`

	// BatchSize is the number of messages in the batch.
	BatchSize int `json:"batch_size,omitempty"`
}

BatchAck is the acknowledgment for a batch publish operation.

func PublishMsgBatch

func PublishMsgBatch(ctx context.Context, js jetstream.JetStream, messages []*nats.Msg, opts ...PublishMsgBatchOpt) (*BatchAck, error)

PublishMsgBatch publishes a batch of messages to a Stream and waits for an ack for the commit.

type BatchFlowControl

type BatchFlowControl struct {
	// AckFirst waits for an ack on the first message in the batch.
	// Default: true
	AckFirst bool

	// AckEvery waits for an ack every N messages (0 = disabled).
	// Default: 0
	AckEvery int

	// AckTimeout is the timeout for waiting for acks when flow control is enabled.
	// Default: timeout from JetStream context.
	AckTimeout time.Duration
}

BatchFlowControl configures flow control for batch publishing.

type BatchMsgOpt

type BatchMsgOpt func(*batchMsgOpts) error

BatchMsgOpt is an option for configuring batch message publishing.

func WithBatchExpectLastSequence

func WithBatchExpectLastSequence(seq uint64) BatchMsgOpt

WithBatchExpectLastSequence sets the expected sequence number the last message on a stream should have. If the last message has a different sequence number server will reject the message and publish will fail.

func WithBatchExpectLastSequenceForSubject

func WithBatchExpectLastSequenceForSubject(seq uint64, subject string) BatchMsgOpt

WithBatchExpectLastSequenceForSubject sets the sequence and subject for which the last sequence number should be checked. If the last message on a subject has a different sequence number server will reject the message and publish will fail.

func WithBatchExpectLastSequencePerSubject

func WithBatchExpectLastSequencePerSubject(seq uint64) BatchMsgOpt

WithBatchExpectLastSequencePerSubject sets the expected sequence number the last message on a subject the message is published to. If the last message on a subject has a different sequence number server will reject the message and publish will fail.

func WithBatchExpectStream

func WithBatchExpectStream(stream string) BatchMsgOpt

WithBatchExpectStream sets the expected stream the message should be published to. If the message is published to a different stream server will reject the message and publish will fail.

func WithBatchMsgTTL

func WithBatchMsgTTL(dur time.Duration) BatchMsgOpt

WithBatchMsgTTL sets per msg TTL for batch messages. Requires [StreamConfig.AllowMsgTTL] to be enabled.

type BatchPublisher

type BatchPublisher interface {
	// Add publishes a message to the batch with the given subject and data.
	// It is an IO operation and the message will be published immediately
	// and persisted upon commit.
	Add(subject string, data []byte, opts ...BatchMsgOpt) error

	// AddMsg publishes a message to the batch.
	AddMsg(msg *nats.Msg, opts ...BatchMsgOpt) error

	// Commit publishes the final message with the given subject and data,
	// and commits the batch. Returns a BatchAck containing the acknowledgment
	// from the server.
	Commit(ctx context.Context, subject string, data []byte, opts ...BatchMsgOpt) (*BatchAck, error)

	// CommitMsg publishes the final message and commits the batch.
	// Returns a BatchAck containing the acknowledgment from the server.
	CommitMsg(ctx context.Context, msg *nats.Msg, opts ...BatchMsgOpt) (*BatchAck, error)

	// Discard cancels the batch without committing.
	// The server will abandon the batch after a timeout.
	Discard() error

	// Size returns the number of messages added to the batch so far.
	Size() int

	// IsClosed returns true if the batch has been committed or discarded.
	IsClosed() bool
}

BatchPublisher provides methods for publishing messages to a stream in batches. Messages are published immediately with batch headers, and the batch is committed with the final message which includes a commit header.

func NewBatchPublisher

func NewBatchPublisher(js jetstream.JetStream, opts ...BatchPublisherOpt) (BatchPublisher, error)

NewBatchPublisher creates a new batch publisher for publishing messages in batches.

type BatchPublisherOpt

type BatchPublisherOpt interface {
	// contains filtered or unexported methods
}

BatchPublisherOpt is a functional option for configuring a BatchPublisher.

type GetBatchOpt

type GetBatchOpt func(*getBatchOpts) error

GetBatchOpt is a function that can be used to configure the behavior of the GetBatch function.

func GetBatchMaxBytes

func GetBatchMaxBytes(maxBytes int) GetBatchOpt

GetBatchMaxBytes sets the maximum number of bytes to fetch. The server will try to fetch messages until the maximum number of bytes is reached (or the batch size is reached).

func GetBatchSeq

func GetBatchSeq(seq uint64) GetBatchOpt

GetBatchSeq sets the sequence number from which to start fetching messages.

func GetBatchStartTime

func GetBatchStartTime(startTime time.Time) GetBatchOpt

GetBatchStartTime sets the start time from which to fetch messages.

func GetBatchSubject

func GetBatchSubject(subj string) GetBatchOpt

GetBatchSubject sets the subject from which to start fetching messages. It may include wildcards.

type GetLastForOpt

type GetLastForOpt func(*getLastBatchOpts) error

GetLastForOpt is a function that can be used to configure the behavior of the GetLastMsgsFor function.

func GetLastMsgsBatchSize

func GetLastMsgsBatchSize(batch int) GetLastForOpt

GetLastMsgsBatchSize sets the optional batch size for fetching messages from multiple subjects.

func GetLastMsgsUpToSeq

func GetLastMsgsUpToSeq(seq uint64) GetLastForOpt

GetLastMsgsUpToSeq sets the sequence number up to which to fetch messages (inclusive).

func GetLastMsgsUpToTime

func GetLastMsgsUpToTime(tm time.Time) GetLastForOpt

GetLastMsgsUpToTime sets the time up to which to fetch messages.

type PublishMsgBatchOpt

type PublishMsgBatchOpt interface {
	// contains filtered or unexported methods
}

PublishMsgBatchOpt is a functional option for configuring PublishMsgBatch.