jetstream

package
v0.0.0-...-3ce476a Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 24, 2024 License: Apache-2.0 Imports: 24 Imported by: 0

README

JetStream Simplified Client

This doc covers the basic usage of the jetstream package in nats.go client.

Overview

jetstream package is a new client API to interact with NATS JetStream, aiming to replace the JetStream client implementation from nats package. The main goal of this package is to provide a simple and clear way to interact with JetStream API. Key differences between jetstream and nats packages include:

  • Using smaller, simpler interfaces to manage streams and consumers
  • Using more granular and predictable approach to consuming messages from a stream, instead of relying on often complicated and unpredictable Subscribe() method (and all of its flavors)
  • Allowing the usage of pull consumers to continuously receive incoming messages (including ordered consumer functionality)
  • Separating JetStream context from core NATS

jetstream package provides several ways of interacting with the API:

  • JetStream - top-level interface, used to create and manage streams, consumers and publishing messages
  • Stream - used to manage consumers for a specific stream, as well as performing stream-specific operations (purging, fetching and deleting messages by sequence number, fetching stream info)
  • Consumer - used to get information about a consumer as well as consuming messages
  • Msg - used for message-specific operations - reading data, headers and metadata, as well as performing various types of acknowledgements

NOTE: jetstream requires nats-server >= 2.9.0 to work correctly.

WARNING: The new API is currently provided as a preview, and will deprecate previous JetStream subscribe APIs. It is encouraged to start experimenting with the new APIs as soon as possible.

Basic usage

package main

import (
    "context"
    "fmt"
    "strconv"
    "time"

    "github.com/matsuev/nats.go"
    "github.com/matsuev/nats.go/jetstream"
)

func main() {
    // In the `jetstream` package, almost all API calls rely on `context.Context` for timeout/cancellation handling
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    nc, _ := nats.Connect(nats.DefaultURL)

    // Create a JetStream management interface
    js, _ := jetstream.New(nc)

    // Create a stream
    s, _ := js.CreateStream(ctx, jetstream.StreamConfig{
        Name:     "ORDERS",
        Subjects: []string{"ORDERS.*"},
    })

    // Publish some messages
    for i := 0; i < 100; i++ {
        js.Publish(ctx, "ORDERS.new", []byte("hello message "+strconv.Itoa(i)))
        fmt.Printf("Published hello message %d\n", i)
    }

    // Create durable consumer
    c, _ := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
        Durable:   "CONS",
        AckPolicy: jetstream.AckExplicitPolicy,
    })

    // Get 10 messages from the consumer
    messageCounter := 0
    msgs, _ := c.Fetch(10)
    for msg := range msgs.Messages() {
        msg.Ack()
        fmt.Printf("Received a JetStream message via fetch: %s\n", string(msg.Data()))
        messageCounter++
    }
    fmt.Printf("received %d messages\n", messageCounter)
    if msgs.Error() != nil {
        fmt.Println("Error during Fetch(): ", msgs.Error())
    }

    // Receive messages continuously in a callback
    cons, _ := c.Consume(func(msg jetstream.Msg) {
        msg.Ack()
        fmt.Printf("Received a JetStream message via callback: %s\n", string(msg.Data()))
        messageCounter++
    })
    defer cons.Stop()

    // Iterate over messages continuously
    it, _ := c.Messages()
    for i := 0; i < 10; i++ {
        msg, _ := it.Next()
        msg.Ack()
        fmt.Printf("Received a JetStream message via iterator: %s\n", string(msg.Data()))
        messageCounter++
    }
    it.Stop()

    // block until all 100 published messages have been processed
    for messageCounter < 100 {
        time.Sleep(10 * time.Millisecond)
    }
}

Streams

jetstream provides methods to manage and list streams, as well as perform stream-specific operations (purging, fetching/deleting messages by sequence id)

Stream management (CRUD)
js, _ := jetstream.New(nc)

// create a stream (this is an idempotent operation)
s, _ := js.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"ORDERS.*"},
})

// update a stream
s, _ = js.UpdateStream(ctx, jetstream.StreamConfig{
    Name:        "ORDERS",
    Subjects:    []string{"ORDERS.*"},
    Description: "updated stream",
})

// get stream handle
s, _ = js.Stream(ctx, "ORDERS")

// delete a stream
js.DeleteStream(ctx, "ORDERS")
Listing streams and stream names
// list streams
streams := js.ListStreams(ctx)
for s := range streams.Info() {
    fmt.Println(s.Config.Name)
}
if streams.Err() != nil {
    fmt.Println("Unexpected error occurred")
}

// list stream names
names := js.StreamNames(ctx)
for name := range names.Name() {
    fmt.Println(name)
}
if names.Err() != nil {
    fmt.Println("Unexpected error occurred")
}
Stream-specific operations

Using Stream interface, it is also possible to:

  • Purge a stream
// remove all messages from a stream
_ = s.Purge(ctx)

// remove all messages from a stream that are stored on a specific subject
_ = s.Purge(ctx, jetstream.WithPurgeSubject("ORDERS.new"))

// remove all messages up to specified sequence number
_ = s.Purge(ctx, jetstream.WithPurgeSequence(100))

// remove messages, but keep 10 newest
_ = s.Purge(ctx, jetstream.WithPurgeKeep(10))
  • Get and messages from stream
// get message from stream with sequence number == 100
msg, _ := s.GetMsg(ctx, 100)

// get last message from "ORDERS.new" subject
msg, _ = s.GetLastMsgForSubject(ctx, "ORDERS.new")

// delete a message with sequence number == 100
_ = s.DeleteMsg(ctx, 100)
  • Get information about a stream
// Fetches latest stream info from server
info, _ := s.Info(ctx)
fmt.Println(info.Config.Name)

// Returns the most recently fetched StreamInfo, without making an API call to the server
cachedInfo := s.CachedInfo()
fmt.Println(cachedInfo.Config.Name)

Consumers

Only pull consumers are supported in jetstream package. However, unlike the JetStream API in nats package, pull consumers allow for continuous message retrieval (similarly to how nats.Subscribe() works). Because of that, push consumers can be easily replaced by pull consumers for most of the use cases.

Consumers management

CRUD operations on consumers can be achieved on 2 levels:

  • on JetStream interface
js, _ := jetstream.New(nc)

// create a consumer (this is an idempotent operation)
// an error will be returned if consumer already exists and has different configuration.
cons, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    Durable: "foo",
    AckPolicy: jetstream.AckExplicitPolicy,
})

// create an ephemeral pull consumer by not providing `Durable`
ephemeral, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    AckPolicy: jetstream.AckExplicitPolicy,
})


// consumer can also be created using CreateOrUpdateConsumer
// this method will either create a consumer if it does not exist
// or update existing consumer (if possible)
cons2 := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    Name: "bar",
})

// consumers can be updated
// an error will be returned if consumer with given name does not exist
// or an illegal property is to be updated (e.g. AckPolicy)
updated, _ := js.UpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    AckPolicy: jetstream.AckExplicitPolicy,
    Description: "updated consumer"
})

// get consumer handle
cons, _ = js.Consumer(ctx, "ORDERS", "foo")

// delete a consumer
js.DeleteConsumer(ctx, "ORDERS", "foo")
  • on Stream interface
// Create a JetStream management interface
js, _ := jetstream.New(nc)

// get stream handle
stream, _ := js.Stream(ctx, "ORDERS")

// create consumer
cons, _ := stream.CreateConsumer(ctx, jetstream.ConsumerConfig{
    Durable:   "foo",
    AckPolicy: jetstream.AckExplicitPolicy,
})

// get consumer handle
cons, _ = stream.Consumer(ctx, "ORDERS", "foo")

// delete a consumer
stream.DeleteConsumer(ctx, "foo")

Consumer interface, returned when creating/fetching consumers, allows fetching ConsumerInfo:

// Fetches latest consumer info from server
info, _ := cons.Info(ctx)
fmt.Println(info.Config.Durable)

// Returns the most recently fetched ConsumerInfo, without making an API call to the server
cachedInfo := cons.CachedInfo()
fmt.Println(cachedInfo.Config.Durable)
Listing consumers and consumer names
// list consumers
consumers := s.ListConsumers(ctx)
for cons := range consumers.Info() {
    fmt.Println(cons.Name)
}
if consumers.Err() != nil {
    fmt.Println("Unexpected error occurred")
}

// list consumer names
names := s.ConsumerNames(ctx)
for name := range names.Name() {
    fmt.Println(name)
}
if names.Err() != nil {
    fmt.Println("Unexpected error occurred")
}
Ordered consumers

jetstream, in addition to basic named/ephemeral consumers, supports ordered consumer functionality. Ordered is strictly processing messages in the order that they were stored on the stream, providing a consistent and deterministic message ordering. It is also resilient to consumer deletion.

Ordered consumers present the same set of message consumption methods as standard pull consumers.

js, _ := jetstream.New(nc)

// create a consumer (this is an idempotent operation)
cons, _ := js.OrderedConsumer(ctx, "ORDERS", jetstream.OrderedConsumerConfig{
    // Filter results from "ORDERS" stream by specific subject
    FilterSubjects: []{"ORDERS.A"},
})
Receiving messages from the consumer

The Consumer interface covers allows fetching messages on demand, with pre-defined batch size on bytes limit, or continuous push-like receiving of messages.

Single fetch

This pattern pattern allows fetching a defined number of messages in a single RPC.

  • Using Fetch or FetchBytes, consumer will return up to the provided number of messages/bytes. By default, Fetch() will wait 30 seconds before timing out (this behavior can be configured using FetchMaxWait() option):
// receive up to 10 messages from the stream
msgs, _ := c.Fetch(10)
for msg := range msgs.Messages() {
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
}
if msgs.Error() != nil {
    // handle error
}

// receive up to 1024 B of data
msgs, _ := c.FetchBytes(1024)
for msg := range msgs.Messages() {
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
}
if msgs.Error() != nil {
    // handle error
}

Similarly, FetchNoWait() can be used in order to only return messages from the stream available at the time of sending request:

// FetchNoWait will not wait for new messages if the whole batch is not available at the time of sending request.
msgs, _ := c.FetchNoWait(10)
for msg := range msgs.Messages() {
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
}
if msgs.Error() != nil {
    // handle error
}

Warning: Both Fetch() and FetchNoWait() have worse performance when used to continuously retrieve messages in comparison to Messages() or Consume() methods, as they do not perform any optimizations (pre-buffering) and new subscription is created for each execution.

Continuous polling

There are 2 ways to achieve push-like behavior using pull consumers in jetstream package. Both Messages() and Consume() methods perform similar optimizations and for most cases can be used interchangeably.

There is an advantage of using Messages() instead of Consume() for work-queue scenarios, where messages should be fetched one by one, as it allows for finer control over fetching single messages on demand.

Subject filtering is achieved by configuring a consumer with a FilterSubject value.

Using Consume() receive messages in a callback
cons, _ := js.CreateOrUpdateConsumer("ORDERS", jetstream.ConsumerConfig{
    AckPolicy: jetstream.AckExplicitPolicy,
    // receive messages from ORDERS.A subject only
    FilterSubject: "ORDERS.A"
}))

consContext, _ := c.Consume(func(msg jetstream.Msg) {
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
})
defer consContext.Stop()

Similarly to Messages(), Consume() can be supplied with options to modify the behavior of a single pull request:

  • PullMaxMessages(int) - up to provided number of messages will be buffered
  • PullMaxBytes(int) - up to provided number of bytes will be buffered. This setting and PullMaxMessages are mutually exclusive
  • PullExpiry(time.Duration) - timeout on a single pull request to the server type PullThresholdMessages int
  • PullThresholdMessages(int) - amount of messages which triggers refilling the buffer
  • PullThresholdBytes(int) - amount of bytes which triggers refilling the buffer
  • PullHeartbeat(time.Duration) - idle heartbeat duration for a single pull request. An error will be triggered if at least 2 heartbeats are missed
  • WithConsumeErrHandler(func (ConsumeContext, error)) - when used, sets a custom error handler on Consume(), allowing e.g. tracking missing heartbeats.

NOTE: Stop() should always be called on ConsumeContext to avoid leaking goroutines.

Using Messages() to iterate over incoming messages
iter, _ := cons.Messages()
for {
    msg, err := iter.Next()
    // Next can return error, e.g. when iterator is closed or no heartbeats were received
    if err != nil {
        //handle error
    }
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
    msg.Ack()
}
iter.Stop()

It can also be configured to only store up to defined number of messages/bytes in the buffer.

// a maximum of 10 messages or 1024 bytes will be stored in memory (whichever is encountered first)
iter, _ := cons.Messages(jetstream.PullMaxMessages(10), jetstream.PullMaxBytes(1024))

Messages() exposes the following options:

  • PullMaxMessages(int) - up to provided number of messages will be buffered
  • PullMaxBytes(int) - up to provided number of bytes will be buffered. This setting and PullMaxMessages are mutually exclusive
  • PullExpiry(time.Duration) - timeout on a single pull request to the server type PullThresholdMessages int
  • PullThresholdMessages(int) - amount of messages which triggers refilling the buffer
  • PullThresholdBytes(int) - amount of bytes which triggers refilling the buffer
  • PullHeartbeat(time.Duration) - idle heartbeat duration for a single pull request. An error will be triggered if at least 2 heartbeats are missed (unless WithMessagesErrOnMissingHeartbeat(false) is used)
Using Messages() to fetch single messages one by one

When implementing work queue, it is possible to use Messages() in order to fetch messages from the server one-by-one, without optimizations and pre-buffering (to avoid redeliveries when processing messages at slow rate).

// PullMaxMessages determines how many messages will be sent to the client in a single pull request
iter, _ := cons.Messages(jetstream.PullMaxMessages(1))
numWorkers := 5
sem := make(chan struct{}, numWorkers)
for {
    sem <- struct{}{}
    go func() {
        defer func() {
            <-sem
        }()
        msg, err := iter.Next()
        if err != nil {
            // handle err
        }
        fmt.Printf("Processing msg: %s\n", string(msg.Data()))
        doWork()
        msg.Ack()
    }()
}

Publishing on stream

JetStream interface allows publishing messages on stream in 2 ways:

Synchronous publish
js, _ := jetstream.New(nc)

// Publish message on subject ORDERS.new
// Given subject has to belong to a stream
ack, err := js.PublishMsg(ctx, &nats.Msg{
    Data:    []byte("hello"),
    Subject: "ORDERS.new",
})
fmt.Printf("Published msg with sequence number %d on stream %q", ack.Sequence, ack.Stream)

// A helper method accepting subject and data as parameters
ack, err = js.Publish(ctx, "ORDERS.new", []byte("hello"))

Both Publish() and PublishMsg() can be supplied with options allowing setting various headers. Additionally, for PublishMsg() headers can be set directly on nats.Msg.

// All 3 implementations are work identically
ack, err := js.PublishMsg(ctx, &nats.Msg{
    Data:    []byte("hello"),
    Subject: "ORDERS.new",
    Header: nats.Header{
        "Nats-Msg-Id": []string{"id"},
    },
})

ack, err = js.PublishMsg(ctx, &nats.Msg{
    Data:    []byte("hello"),
    Subject: "ORDERS.new",
}, jetstream.WithMsgID("id"))

ack, err = js.Publish(ctx, "ORDERS.new", []byte("hello"), jetstream.WithMsgID("id"))
Async publish
js, _ := jetstream.New(nc)

// publish message and do not wait for ack
ackF, err := js.PublishMsgAsync(ctx, &nats.Msg{
    Data:    []byte("hello"),
    Subject: "ORDERS.new",
})

// block and wait for ack
select {
case ack := <-ackF.Ok():
    fmt.Printf("Published msg with sequence number %d on stream %q", ack.Sequence, ack.Stream)
case err := <-ackF.Err():
    fmt.Println(err)
}

// similarly to synchronous publish, there is a helper method accepting subject and data
ackF, err = js.PublishAsync("ORDERS.new", []byte("hello"))

Just as for synchronous publish, PublishAsync() and PublishMsgAsync() accept options for setting headers.

Examples

You can find more examples of jetstream usage here.

Documentation

Index

Constants

View Source 
const (
	KeyValueMaxHistory = 64
	AllKeys            = ">"
)
View Source 
const (
	MsgIDHeader               = "Nats-Msg-Id"
	ExpectedStreamHeader      = "Nats-Expected-Stream"
	ExpectedLastSeqHeader     = "Nats-Expected-Last-Sequence"
	ExpectedLastSubjSeqHeader = "Nats-Expected-Last-Subject-Sequence"
	ExpectedLastMsgIDHeader   = "Nats-Expected-Last-Msg-Id"
	MsgRollup                 = "Nats-Rollup"
)
View Source 
const (
	StreamHeader       = "Nats-Stream"
	SequenceHeader     = "Nats-Sequence"
	TimeStampHeaer     = "Nats-Time-Stamp"
	SubjectHeader      = "Nats-Subject"
	LastSequenceHeader = "Nats-Last-Sequence"
)

Headers for republished messages and direct gets.

View Source 
const (
	MsgRollupSubject = "sub"
	MsgRollupAll     = "all"
)

Rollups, can be subject only or all messages.

View Source 
const (
	// Default time wait between retries on Publish if err is NoResponders.
	DefaultPubRetryWait = 250 * time.Millisecond

	// Default number of retries
	DefaultPubRetryAttempts = 2
)
View Source 
const (
	DefaultMaxMessages = 500
	DefaultExpires     = 30 * time.Second
)
View Source 
const (
	// DefaultAPIPrefix is the default prefix for the JetStream API.
	DefaultAPIPrefix = "$JS.API."
)

Request API subjects for JetStream.

Variables

View Source 
var (

	// ErrJetStreamNotEnabled is an error returned when JetStream is not enabled.
	ErrJetStreamNotEnabled JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeJetStreamNotEnabled, Description: "jetstream not enabled", Code: 503}}

	// ErrJetStreamNotEnabledForAccount is an error returned when JetStream is not enabled for an account.
	ErrJetStreamNotEnabledForAccount JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeJetStreamNotEnabledForAccount, Description: "jetstream not enabled for account", Code: 503}}

	// ErrStreamNotFound is an error returned when stream with given name does not exist.
	ErrStreamNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNotFound, Description: "stream not found", Code: 404}}

	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration.
	ErrStreamNameAlreadyInUse JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNameInUse, Description: "stream name already in use", Code: 400}}

	// ErrStreamSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
	// the stream subject transform. If this error is returned when executing CreateStream(), the stream with invalid
	// configuration was already created in the server.
	ErrStreamSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}

	// ErrStreamSourceSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
	// the stream source subject transform. If this error is returned when executing CreateStream(), the stream with invalid
	// configuration was already created in the server.
	ErrStreamSourceSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}

	// ErrStreamSourceNotSupported is returned when the connected nats-server version does not support setting
	// the stream sources. If this error is returned when executing CreateStream(), the stream with invalid
	// configuration was already created in the server.
	ErrStreamSourceNotSupported JetStreamError = &jsError{message: "stream sourcing is not supported by nats-server"}

	// ErrStreamSourceMultipleFilterSubjectsNotSupported is returned when the connected nats-server version does not support setting
	// the stream sources. If this error is returned when executing CreateStream(), the stream with invalid
	// configuration was already created in the server.
	ErrStreamSourceMultipleFilterSubjectsNotSupported JetStreamError = &jsError{message: "stream sourcing with multiple subject filters not supported by nats-server"}

	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}

	// ErrConsumerExists is returned when attempting to create a consumer with CreateConsumer but a consumer with given name already exists.
	ErrConsumerExists JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerExists, Description: "consumer already exists", Code: 400}}

	// ErrConsumerNameExists is returned when attempting to update a consumer with UpdateConsumer but a consumer with given name does not exist.
	ErrConsumerDoesNotExist JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerDoesNotExist, Description: "consumer does not exist", Code: 400}}

	// ErrMsgNotFound is returned when message with provided sequence number does not exist.
	ErrMsgNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeMessageNotFound, Description: "message not found", Code: 404}}

	// ErrBadRequest is returned when invalid request is sent to JetStream API.
	ErrBadRequest JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeBadRequest, Description: "bad request", Code: 400}}

	// ErrConsumerCreate is returned when nats-server reports error when creating consumer (e.g. illegal update).
	ErrConsumerCreate JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerCreate, Description: "could not create consumer", Code: 500}}

	// ErrDuplicateFilterSubjects is returned when both FilterSubject and FilterSubjects are specified when creating consumer.
	ErrDuplicateFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeDuplicateFilterSubjects, Description: "consumer cannot have both FilterSubject and FilterSubjects specified", Code: 500}}

	// ErrDuplicateFilterSubjects is returned when filter subjects overlap when creating consumer.
	ErrOverlappingFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeOverlappingFilterSubjects, Description: "consumer subject filters cannot overlap", Code: 500}}

	// ErrEmptyFilter is returned when a filter in FilterSubjects is empty.
	ErrEmptyFilter JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerEmptyFilter, Description: "consumer filter in FilterSubjects cannot be empty", Code: 500}}

	// ErrConsumerMultipleFilterSubjectsNotSupported is returned when the connected nats-server version does not support setting
	// multiple filter subjects with filter_subjects field. If this error is returned when executing AddConsumer(), the consumer with invalid
	// configuration was already created in the server.
	ErrConsumerMultipleFilterSubjectsNotSupported JetStreamError = &jsError{message: "multiple consumer filter subjects not supported by nats-server"}

	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
	ErrConsumerNameAlreadyInUse JetStreamError = &jsError{message: "consumer name already in use"}

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

	// ErrStreamNameRequired is returned when the provided stream name is empty.
	ErrStreamNameRequired JetStreamError = &jsError{message: "stream name is required"}

	// ErrMsgAlreadyAckd is returned when attempting to acknowledge message more than once.
	ErrMsgAlreadyAckd JetStreamError = &jsError{message: "message was already acknowledged"}

	// ErrNoStreamResponse is returned when there is no response from stream (e.g. no responders error).
	ErrNoStreamResponse JetStreamError = &jsError{message: "no response from stream"}

	// ErrNotJSMessage is returned when attempting to get metadata from non JetStream message.
	ErrNotJSMessage JetStreamError = &jsError{message: "not a jetstream message"}

	// ErrInvalidStreamName is returned when the provided stream name is invalid (contains '.').
	ErrInvalidStreamName JetStreamError = &jsError{message: "invalid stream name"}

	// ErrInvalidSubject is returned when the provided subject name is invalid.
	ErrInvalidSubject JetStreamError = &jsError{message: "invalid subject name"}

	// ErrInvalidConsumerName is returned when the provided consumer name is invalid (contains '.').
	ErrInvalidConsumerName JetStreamError = &jsError{message: "invalid consumer name"}

	// ErrNoMessages is returned when no messages are currently available for a consumer.
	ErrNoMessages = &jsError{message: "no messages"}

	// ErrMaxBytesExceeded is returned when a message would exceed MaxBytes set on a pull request.
	ErrMaxBytesExceeded = &jsError{message: "message size exceeds max bytes"}

	// ErrConsumerDeleted is returned when attempting to send pull request to a consumer which does not exist.
	ErrConsumerDeleted JetStreamError = &jsError{message: "consumer deleted"}

	// ErrConsumerLeadershipChanged is returned when pending requests are no longer valid after leadership has changed.
	ErrConsumerLeadershipChanged JetStreamError = &jsError{message: "leadership change"}

	// ErrHandlerRequired is returned when no handler func is provided in Stream().
	ErrHandlerRequired = &jsError{message: "handler cannot be empty"}

	// ErrEndOfData is returned when iterating over paged API from JetStream reaches end of data.
	ErrEndOfData = &jsError{message: "end of data reached"}

	// ErrNoHeartbeat is received when no message is received in IdleHeartbeat time (if set).
	ErrNoHeartbeat = &jsError{message: "no heartbeat received"}

	// ErrConsumerHasActiveSubscription is returned when a consumer is already subscribed to a stream.
	ErrConsumerHasActiveSubscription = &jsError{message: "consumer has active subscription"}

	// ErrMsgNotBound is returned when given message is not bound to any subscription.
	ErrMsgNotBound = &jsError{message: "message is not bound to subscription/connection"}

	// ErrMsgNoReply is returned when attempting to reply to a message without a reply subject.
	ErrMsgNoReply = &jsError{message: "message does not have a reply"}

	// ErrMsgDeleteUnsuccessful is returned when an attempt to delete a message is unsuccessful.
	ErrMsgDeleteUnsuccessful = &jsError{message: "message deletion unsuccessful"}

	// ErrAsyncPublishReplySubjectSet is returned when reply subject is set on async message publish.
	ErrAsyncPublishReplySubjectSet = &jsError{message: "reply subject should be empty"}

	// ErrTooManyStalledMsgs is returned when too many outstanding async messages are waiting for ack.
	ErrTooManyStalledMsgs = &jsError{message: "stalled with too many outstanding async published messages"}

	// ErrInvalidOption is returned when there is a collision between options.
	ErrInvalidOption = &jsError{message: "invalid jetstream option"}

	// ErrMsgIteratorClosed is returned when attempting to get message from a closed iterator.
	ErrMsgIteratorClosed = &jsError{message: "messages iterator closed"}

	// ErrOrderedConsumerReset is returned when resetting ordered consumer fails due to too many attempts.
	ErrOrderedConsumerReset = &jsError{message: "recreating ordered consumer"}

	// ErrOrderConsumerUsedAsFetch is returned when ordered consumer was already used to process
	// messages using Fetch (or FetchBytes).
	ErrOrderConsumerUsedAsFetch = &jsError{message: "ordered consumer initialized as fetch"}

	// ErrOrderConsumerUsedAsConsume is returned when ordered consumer was already used to process
	// messages using Consume or Messages.
	ErrOrderConsumerUsedAsConsume = &jsError{message: "ordered consumer initialized as consume"}

	// ErrOrderedConsumerConcurrentRequests is returned when attempting to run concurrent operations
	// on ordered consumers.
	ErrOrderedConsumerConcurrentRequests = &jsError{message: "cannot run concurrent processing using ordered consumer"}

	// ErrOrderedConsumerNotCreated is returned when trying to get consumer info of an
	// ordered consumer which was not yet created.
	ErrOrderedConsumerNotCreated = &jsError{message: "consumer instance not yet created"}

	// ErrKeyExists is returned when attempting to create a key that already exists.
	ErrKeyExists JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamWrongLastSequence, Code: 400}, message: "key exists"}

	// ErrKeyValueConfigRequired is returned when attempting to create a bucket without a config.
	ErrKeyValueConfigRequired = &jsError{message: "config required"}

	// ErrInvalidBucketName is returned when attempting to create a bucket with an invalid name.
	ErrInvalidBucketName = &jsError{message: "invalid bucket name"}

	// ErrInvalidKey is returned when attempting to create a key with an invalid name.
	ErrInvalidKey = &jsError{message: "invalid key"}

	// ErrBucketNotFound is returned when attempting to access a bucket that does not exist.
	ErrBucketNotFound = &jsError{message: "bucket not found"}

	// ErrBadBucket is returned when attempting to access a bucket that is not a key-value store.
	ErrBadBucket = &jsError{message: "bucket not valid key-value store"}

	// ErrKeyNotFound is returned when attempting to access a key that does not exist.
	ErrKeyNotFound = &jsError{message: "key not found"}

	// ErrKeyDeleted is returned when attempting to access a key that was deleted.
	ErrKeyDeleted = &jsError{message: "key was deleted"}

	// ErrHistoryToLarge is returned when provided history limit is larger than 64.
	ErrHistoryTooLarge = &jsError{message: "history limited to a max of 64"}

	// ErrNoKeysFound is returned when no keys are found.
	ErrNoKeysFound = &jsError{message: "no keys found"}

	// ErrObjectConfigRequired is returned when attempting to create an object without a config.
	ErrObjectConfigRequired = &jsError{message: "object-store config required"}

	// ErrBadObjectMeta is returned when the meta information of an object is invalid.
	ErrBadObjectMeta = &jsError{message: "object-store meta information invalid"}

	// ErrObjectNotFound is returned when an object is not found.
	ErrObjectNotFound = &jsError{message: "object not found"}

	// ErrInvalidStoreName is returned when the name of an object-store is invalid.
	ErrInvalidStoreName = &jsError{message: "invalid object-store name"}

	// ErrDigestMismatch is returned when the digests of an object do not match.
	ErrDigestMismatch = &jsError{message: "received a corrupt object, digests do not match"}

	// ErrInvalidDigestFormat is returned when the digest hash of an object has an invalid format.
	ErrInvalidDigestFormat = &jsError{message: "object digest hash has invalid format"}

	// ErrNoObjectsFound is returned when no objects are found.
	ErrNoObjectsFound = &jsError{message: "no objects found"}

	// ErrObjectAlreadyExists is returned when an object with the same name already exists.
	ErrObjectAlreadyExists = &jsError{message: "an object already exists with that name"}

	// ErrNameRequired is returned when a name is required.
	ErrNameRequired = &jsError{message: "name is required"}

	// ErrLinkNotAllowed is returned when a link cannot be set when putting the object in a bucket.
	ErrLinkNotAllowed = &jsError{message: "link cannot be set when putting the object in bucket"}

	// ErrObjectRequired is returned when an object is required.
	ErrObjectRequired = &jsError{message: "object required"}

	// ErrNoLinkToDeleted is returned when it is not allowed to link to a deleted object.
	ErrNoLinkToDeleted = &jsError{message: "not allowed to link to a deleted object"}

	// ErrNoLinkToLink is returned when it is not allowed to link to another link.
	ErrNoLinkToLink = &jsError{message: "not allowed to link to another link"}

	// ErrCantGetBucket is returned when an invalid Get is attempted on an object that is a link to a bucket.
	ErrCantGetBucket = &jsError{message: "invalid Get, object is a link to a bucket"}

	// ErrBucketRequired is returned when a bucket is required.
	ErrBucketRequired = &jsError{message: "bucket required"}

	// ErrBucketMalformed is returned when a bucket is malformed.
	ErrBucketMalformed = &jsError{message: "bucket malformed"}

	// ErrUpdateMetaDeleted is returned when the meta information of a deleted object cannot be updated.
	ErrUpdateMetaDeleted = &jsError{message: "cannot update meta for a deleted object"}
)

Functions

func DecodeObjectDigest 

func DecodeObjectDigest(data string) ([]byte, error)

DecodeObjectDigest decodes base64 hash

func GetObjectDigestValue 

func GetObjectDigestValue(data hash.Hash) string

GetObjectDigestValue calculates the base64 value of hashed data

Types

type APIError 

type APIError struct {
	Code        int       `json:"code"`
	ErrorCode   ErrorCode `json:"err_code"`
	Description string    `json:"description,omitempty"`
}

APIError is included in all API responses if there was an error.

func (*APIError) APIError 

func (e *APIError) APIError() *APIError

APIError implements the JetStreamError interface.

func (*APIError) Error 

func (e *APIError) Error() string

Error prints the JetStream API error code and description

func (*APIError) Is 

func (e *APIError) Is(err error) bool

Is matches against an APIError.

type APIStats 

type APIStats struct {
	Total  uint64 `json:"total"`
	Errors uint64 `json:"errors"`
}

APIStats reports on API calls to JetStream for this account.

type AccountInfo 

type AccountInfo struct {
	Tier
	Domain string          `json:"domain"`
	API    APIStats        `json:"api"`
	Tiers  map[string]Tier `json:"tiers"`
}

AccountInfo contains info about the JetStream usage from the current account.

type AccountLimits 

type AccountLimits struct {
	MaxMemory    int64 `json:"max_memory"`
	MaxStore     int64 `json:"max_storage"`
	MaxStreams   int   `json:"max_streams"`
	MaxConsumers int   `json:"max_consumers"`
}

AccountLimits includes the JetStream limits of the current account.

type AckPolicy 

type AckPolicy int

AckPolicy determines how the consumer should acknowledge delivered messages. 

const (
	// AckExplicitPolicy requires ack or nack for all messages.
	AckExplicitPolicy AckPolicy = iota

	// AckAllPolicy when acking a sequence number, this implicitly acks all
	// sequences below this one as well.
	AckAllPolicy

	// AckNonePolicy requires no acks for delivered messages.
	AckNonePolicy
)

func (AckPolicy) MarshalJSON 

func (p AckPolicy) MarshalJSON() ([]byte, error)

func (AckPolicy) String 

func (p AckPolicy) String() string

func (*AckPolicy) UnmarshalJSON 

func (p *AckPolicy) UnmarshalJSON(data []byte) error

type ClientTrace 

type ClientTrace struct {
	RequestSent      func(subj string, payload []byte)
	ResponseReceived func(subj string, payload []byte, hdr nats.Header)
}

ClientTrace can be used to trace API interactions for the JetStream Context.

type ClusterInfo 

type ClusterInfo struct {
	Name     string      `json:"name,omitempty"`
	Leader   string      `json:"leader,omitempty"`
	Replicas []*PeerInfo `json:"replicas,omitempty"`
}

ClusterInfo shows information about the underlying set of servers that make up the stream or consumer.

type ConsumeContext 

type ConsumeContext interface {
	// Stop unsubscribes from the stream and cancels subscription.
	// No more messages will be received after calling this method.
	// All messages that are already in the buffer are discarded.
	Stop()

	// Drain unsubscribes from the stream and cancels subscription.
	// All messages that are already in the buffer will be processed in callback function.
	Drain()
}

type ConsumeErrHandlerFunc 

type ConsumeErrHandlerFunc func(consumeCtx ConsumeContext, err error)

type Consumer 

type Consumer interface {
	// Fetch is used to retrieve up to a provided number of messages from a stream.
	// This method will always send a single request and wait until either all messages are retrieved
	// or request times out.
	Fetch(batch int, opts ...FetchOpt) (MessageBatch, error)
	// FetchBytes is used to retrieve up to a provided bytes from the stream.
	// This method will always send a single request and wait until provided number of bytes is
	// exceeded or request times out.
	FetchBytes(maxBytes int, opts ...FetchOpt) (MessageBatch, error)
	// FetchNoWait is used to retrieve up to a provided number of messages from a stream.
	// This method will always send a single request and immediately return up to a provided number of messages.
	FetchNoWait(batch int) (MessageBatch, error)
	// Consume can be used to continuously receive messages and handle them with the provided callback function
	Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error)
	// Messages returns [MessagesContext], allowing continuously iterating over messages on a stream.
	Messages(opts ...PullMessagesOpt) (MessagesContext, error)
	// Next is used to retrieve the next message from the stream.
	// This method will block until the message is retrieved or timeout is reached.
	Next(opts ...FetchOpt) (Msg, error)

	// Info returns Consumer details
	Info(context.Context) (*ConsumerInfo, error)
	// CachedInfo returns [*ConsumerInfo] cached on a consumer struct
	CachedInfo() *ConsumerInfo
}

Consumer contains methods for fetching/processing messages from a stream, as well as fetching consumer info

type ConsumerConfig 

type ConsumerConfig struct {
	Name               string          `json:"name,omitempty"`
	Durable            string          `json:"durable_name,omitempty"`
	Description        string          `json:"description,omitempty"`
	DeliverPolicy      DeliverPolicy   `json:"deliver_policy"`
	OptStartSeq        uint64          `json:"opt_start_seq,omitempty"`
	OptStartTime       *time.Time      `json:"opt_start_time,omitempty"`
	AckPolicy          AckPolicy       `json:"ack_policy"`
	AckWait            time.Duration   `json:"ack_wait,omitempty"`
	MaxDeliver         int             `json:"max_deliver,omitempty"`
	BackOff            []time.Duration `json:"backoff,omitempty"`
	FilterSubject      string          `json:"filter_subject,omitempty"`
	ReplayPolicy       ReplayPolicy    `json:"replay_policy"`
	RateLimit          uint64          `json:"rate_limit_bps,omitempty"` // Bits per sec
	SampleFrequency    string          `json:"sample_freq,omitempty"`
	MaxWaiting         int             `json:"max_waiting,omitempty"`
	MaxAckPending      int             `json:"max_ack_pending,omitempty"`
	HeadersOnly        bool            `json:"headers_only,omitempty"`
	MaxRequestBatch    int             `json:"max_batch,omitempty"`
	MaxRequestExpires  time.Duration   `json:"max_expires,omitempty"`
	MaxRequestMaxBytes int             `json:"max_bytes,omitempty"`

	// Inactivity threshold.
	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`

	// Generally inherited by parent stream and other markers, now can be configured directly.
	Replicas int `json:"num_replicas"`
	// Force memory storage.
	MemoryStorage bool `json:"mem_storage,omitempty"`

	// NOTE: FilterSubjects requires nats-server v2.10.0+
	FilterSubjects []string `json:"filter_subjects,omitempty"`

	// Metadata is additional metadata for the Consumer.
	// Keys starting with `_nats` are reserved.
	// NOTE: Metadata requires nats-server v2.10.0+
	Metadata map[string]string `json:"metadata,omitempty"`
}

ConsumerConfig is the configuration of a JetStream consumer.

type ConsumerInfo 

type ConsumerInfo struct {
	Stream         string         `json:"stream_name"`
	Name           string         `json:"name"`
	Created        time.Time      `json:"created"`
	Config         ConsumerConfig `json:"config"`
	Delivered      SequenceInfo   `json:"delivered"`
	AckFloor       SequenceInfo   `json:"ack_floor"`
	NumAckPending  int            `json:"num_ack_pending"`
	NumRedelivered int            `json:"num_redelivered"`
	NumWaiting     int            `json:"num_waiting"`
	NumPending     uint64         `json:"num_pending"`
	Cluster        *ClusterInfo   `json:"cluster,omitempty"`
	PushBound      bool           `json:"push_bound,omitempty"`
}

ConsumerInfo is the info from a JetStream consumer.

type ConsumerInfoLister 

type ConsumerInfoLister interface {
	Info() <-chan *ConsumerInfo
	Err() error
}

type ConsumerNameLister 

type ConsumerNameLister interface {
	Name() <-chan string
	Err() error
}

type DeleteMarkersOlderThan 

type DeleteMarkersOlderThan time.Duration

DeleteMarkersOlderThan indicates that delete or purge markers older than that will be deleted as part of PurgeDeletes() operation, otherwise, only the data will be removed but markers that are recent will be kept. Note that if no option is specified, the default is 30 minutes. You can set this option to a negative value to instruct to always remove the markers, regardless of their age.

type DeliverPolicy 

type DeliverPolicy int
const (
	// DeliverAllPolicy starts delivering messages from the very beginning of a
	// stream. This is the default.
	DeliverAllPolicy DeliverPolicy = iota

	// DeliverLastPolicy will start the consumer with the last sequence
	// received.
	DeliverLastPolicy

	// DeliverNewPolicy will only deliver new messages that are sent after the
	// consumer is created.
	DeliverNewPolicy

	// DeliverByStartSequencePolicy will deliver messages starting from a given
	// sequence.
	DeliverByStartSequencePolicy

	// DeliverByStartTimePolicy will deliver messages starting from a given
	// time.
	DeliverByStartTimePolicy

	// DeliverLastPerSubjectPolicy will start the consumer with the last message
	// for all subjects received.
	DeliverLastPerSubjectPolicy
)

func (DeliverPolicy) MarshalJSON 

func (p DeliverPolicy) MarshalJSON() ([]byte, error)

func (DeliverPolicy) String 

func (p DeliverPolicy) String() string

func (*DeliverPolicy) UnmarshalJSON 

func (p *DeliverPolicy) UnmarshalJSON(data []byte) error

type DiscardPolicy 

type DiscardPolicy int

DiscardPolicy determines how to proceed when limits of messages or bytes are reached. 

const (
	// DiscardOld will remove older messages to return to the limits. This is
	// the default.
	DiscardOld DiscardPolicy = iota
	// DiscardNew will fail to store new messages.
	DiscardNew
)

func (DiscardPolicy) MarshalJSON 

func (dp DiscardPolicy) MarshalJSON() ([]byte, error)

func (DiscardPolicy) String 

func (dp DiscardPolicy) String() string

func (*DiscardPolicy) UnmarshalJSON 

func (dp *DiscardPolicy) UnmarshalJSON(data []byte) error

type ErrorCode 

type ErrorCode uint16

ErrorCode represents error_code returned in response from JetStream API 

const (
	JSErrCodeJetStreamNotEnabledForAccount ErrorCode = 10039
	JSErrCodeJetStreamNotEnabled           ErrorCode = 10076

	JSErrCodeStreamNotFound  ErrorCode = 10059
	JSErrCodeStreamNameInUse ErrorCode = 10058

	JSErrCodeConsumerCreate            ErrorCode = 10012
	JSErrCodeConsumerNotFound          ErrorCode = 10014
	JSErrCodeConsumerNameExists        ErrorCode = 10013
	JSErrCodeConsumerAlreadyExists     ErrorCode = 10105
	JSErrCodeConsumerExists            ErrorCode = 10148
	JSErrCodeDuplicateFilterSubjects   ErrorCode = 10136
	JSErrCodeOverlappingFilterSubjects ErrorCode = 10138
	JSErrCodeConsumerEmptyFilter       ErrorCode = 10139
	JSErrCodeConsumerDoesNotExist      ErrorCode = 10149

	JSErrCodeMessageNotFound ErrorCode = 10037

	JSErrCodeBadRequest ErrorCode = 10003

	JSErrCodeStreamWrongLastSequence ErrorCode = 10071
)

type ExternalStream 

type ExternalStream struct {
	APIPrefix     string `json:"api"`
	DeliverPrefix string `json:"deliver"`
}

ExternalStream allows you to qualify access to a stream source in another account.

type FetchOpt 

type FetchOpt func(*pullRequest) error

func FetchMaxWait 

func FetchMaxWait(timeout time.Duration) FetchOpt

FetchMaxWait sets custom timeout for fetching predefined batch of messages

type GetMsgOpt 

type GetMsgOpt func(*apiMsgGetRequest) error

func WithGetMsgSubject 

func WithGetMsgSubject(subject string) GetMsgOpt

WithGetMsgSubject sets the stream subject from which the message should be retrieved. Server will return a first message with a seq >= to the input seq that has the specified subject.

type GetObjectInfoOpt 

type GetObjectInfoOpt func(opts *getObjectInfoOpts) error

func GetObjectInfoShowDeleted 

func GetObjectInfoShowDeleted() GetObjectInfoOpt

GetObjectInfoShowDeleted makes GetInfo() return object if it was marked as deleted.

type GetObjectOpt 

type GetObjectOpt func(opts *getObjectOpts) error

func GetObjectShowDeleted 

func GetObjectShowDeleted() GetObjectOpt

GetObjectShowDeleted makes Get() return object if it was marked as deleted.

type JetStream 

type JetStream interface {
	// Returns *AccountInfo, containing details about the account associated with this JetStream connection
	AccountInfo(ctx context.Context) (*AccountInfo, error)

	StreamConsumerManager
	StreamManager
	Publisher
	KeyValueManager
	ObjectStoreManager
}

JetStream contains CRUD methods to operate on a stream Create, update and get operations return 'Stream' interface, allowing operations on consumers

CreateOrUpdateConsumer, Consumer and DeleteConsumer are helper methods used to create/fetch/remove consumer without fetching stream (bypassing stream API)

Client returns a JetStremClient, used to publish messages on a stream or fetch messages by sequence number

func New 

func New(nc *nats.Conn, opts ...JetStreamOpt) (JetStream, error)

New returns a new JetStream instance.

Available options: WithClientTrace - enables request/response tracing. WithPublishAsyncErrHandler - sets error handler for async message publish. WithPublishAsyncMaxPending - sets the maximum outstanding async publishes that can be inflight at one time. [WithDirectGet] - specifies whether client should use direct get requests.

func NewWithAPIPrefix 

func NewWithAPIPrefix(nc *nats.Conn, apiPrefix string, opts ...JetStreamOpt) (JetStream, error)

NewWithAPIPrefix returns a new JetStream instance and sets the API prefix to be used in requests to JetStream API

Available options: WithClientTrace - enables request/response tracing WithPublishAsyncErrHandler - sets error handler for async message publish WithPublishAsyncMaxPending - sets the maximum outstanding async publishes that can be inflight at one time. [WithDirectGet] - specifies whether client should use direct get requests.

func NewWithDomain 

func NewWithDomain(nc *nats.Conn, domain string, opts ...JetStreamOpt) (JetStream, error)

NewWithDomain returns a new JetStream instance and sets the domain name token used when sending JetStream requests

Available options: WithClientTrace - enables request/response tracing WithPublishAsyncErrHandler - sets error handler for async message publish WithPublishAsyncMaxPending - sets the maximum outstanding async publishes that can be inflight at one time. [WithDirectGet] - specifies whether client should use direct get requests.

type JetStreamError 

type JetStreamError interface {
	APIError() *APIError
	error
}

JetStreamError is an error result that happens when using JetStream. In case of client-side error, APIError returns nil

type JetStreamOpt 

type JetStreamOpt func(*jsOpts) error

func WithClientTrace 

func WithClientTrace(ct *ClientTrace) JetStreamOpt

WithClientTrace enables request/response API calls tracing ClientTrace is used to provide handlers for each event

func WithPublishAsyncErrHandler 

func WithPublishAsyncErrHandler(cb MsgErrHandler) JetStreamOpt

WithPublishAsyncErrHandler sets error handler for async message publish

func WithPublishAsyncMaxPending 

func WithPublishAsyncMaxPending(max int) JetStreamOpt

WithPublishAsyncMaxPending sets the maximum outstanding async publishes that can be inflight at one time.

type KVDeleteOpt 

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

func LastRevision 

func LastRevision(revision uint64) KVDeleteOpt

LastRevision deletes if the latest revision matches.

type KVPurgeOpt 

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

type KeyLister 

type KeyLister interface {
	Keys() <-chan string
	Stop() error
}

KeyLister is used to retrieve a list of key value store keys

type KeyValue 

type KeyValue interface {
	// Get returns the latest value for the key.
	Get(ctx context.Context, key string) (KeyValueEntry, error)
	// GetRevision returns a specific revision value for the key.
	GetRevision(ctx context.Context, key string, revision uint64) (KeyValueEntry, error)
	// Put will place the new value for the key into the store.
	Put(ctx context.Context, key string, value []byte) (uint64, error)
	// PutString will place the string for the key into the store.
	PutString(ctx context.Context, key string, value string) (uint64, error)
	// Create will add the key/value pair if it does not exist.
	Create(ctx context.Context, key string, value []byte) (uint64, error)
	// Update will update the value if the latest revision matches.
	Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
	// Delete will place a delete marker and leave all revisions.
	Delete(ctx context.Context, key string, opts ...KVDeleteOpt) error
	// Purge will place a delete marker and remove all previous revisions.
	Purge(ctx context.Context, key string, opts ...KVDeleteOpt) error
	// Watch for any updates to keys that match the keys argument which could include wildcards.
	// Watch will send a nil entry when it has received all initial values.
	Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWatcher, error)
	// WatchAll will invoke the callback for all updates.
	WatchAll(ctx context.Context, opts ...WatchOpt) (KeyWatcher, error)
	// Keys will return all keys.
	// DEPRECATED: Use ListKeys instead to avoid memory issues.
	Keys(ctx context.Context, opts ...WatchOpt) ([]string, error)
	// ListKeys will return all keys in a channel.
	ListKeys(ctx context.Context, opts ...WatchOpt) (KeyLister, error)
	// History will return all historical values for the key.
	History(ctx context.Context, key string, opts ...WatchOpt) ([]KeyValueEntry, error)
	// Bucket returns the current bucket name.
	Bucket() string
	// PurgeDeletes will remove all current delete markers.
	PurgeDeletes(ctx context.Context, opts ...KVPurgeOpt) error
	// Status retrieves the status and configuration of a bucket
	Status(ctx context.Context) (KeyValueStatus, error)
}

KeyValue contains methods to operate on a KeyValue store.

type KeyValueBucketStatus 

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

KeyValueBucketStatus represents status of a Bucket, implements KeyValueStatus

func (*KeyValueBucketStatus) BackingStore 

func (s *KeyValueBucketStatus) BackingStore() string

BackingStore indicates what technology is used for storage of the bucket

func (*KeyValueBucketStatus) Bucket 

func (s *KeyValueBucketStatus) Bucket() string

Bucket the name of the bucket

func (*KeyValueBucketStatus) Bytes 

func (s *KeyValueBucketStatus) Bytes() uint64

Bytes is the size of the stream

func (*KeyValueBucketStatus) History 

func (s *KeyValueBucketStatus) History() int64

History returns the configured history kept per key

func (*KeyValueBucketStatus) IsCompressed 

func (s *KeyValueBucketStatus) IsCompressed() bool

IsCompressed indicates if the data is compressed on disk

func (*KeyValueBucketStatus) StreamInfo 

func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo

StreamInfo is the stream info retrieved to create the status

func (*KeyValueBucketStatus) TTL 

func (s *KeyValueBucketStatus) TTL() time.Duration

TTL is how long the bucket keeps values for

func (*KeyValueBucketStatus) Values 

func (s *KeyValueBucketStatus) Values() uint64

Values is how many messages are in the bucket, including historical values

type KeyValueConfig 

type KeyValueConfig struct {
	Bucket       string
	Description  string
	MaxValueSize int32
	History      uint8
	TTL          time.Duration
	MaxBytes     int64
	Storage      StorageType
	Replicas     int
	Placement    *Placement
	RePublish    *RePublish
	Mirror       *StreamSource
	Sources      []*StreamSource

	// Enable underlying stream compression.
	// NOTE: Compression is supported for nats-server 2.10.0+
	Compression bool
}

KeyValueConfig is for configuring a KeyValue store.

type KeyValueEntry 

type KeyValueEntry interface {
	// Bucket is the bucket the data was loaded from.
	Bucket() string
	// Key is the key that was retrieved.
	Key() string
	// Value is the retrieved value.
	Value() []byte
	// Revision is a unique sequence for this value.
	Revision() uint64
	// Created is the time the data was put in the bucket.
	Created() time.Time
	// Delta is distance from the latest value.
	Delta() uint64
	// Operation returns Put or Delete or Purge.
	Operation() KeyValueOp
}

KeyValueEntry is a retrieved entry for Get or List or Watch.

type KeyValueLister 

type KeyValueLister interface {
	Status() <-chan KeyValueStatus
	Error() error
}

Public interfaces and structs

type KeyValueManager 

type KeyValueManager interface {
	// KeyValue will lookup and bind to an existing KeyValue store.
	KeyValue(ctx context.Context, bucket string) (KeyValue, error)
	// CreateKeyValue will create a KeyValue store with the following configuration.
	CreateKeyValue(ctx context.Context, cfg KeyValueConfig) (KeyValue, error)
	// DeleteKeyValue will delete this KeyValue store (JetStream stream).
	DeleteKeyValue(ctx context.Context, bucket string) error
	// KeyValueStoreNames is used to retrieve a list of key value store names
	KeyValueStoreNames(ctx context.Context) KeyValueNamesLister
	// KeyValueStores is used to retrieve a list of key value store statuses
	KeyValueStores(ctx context.Context) KeyValueLister
}

KeyValueManager is used to manage KeyValue stores.

type KeyValueNamesLister 

type KeyValueNamesLister interface {
	Name() <-chan string
	Error() error
}

Public interfaces and structs

type KeyValueOp 

type KeyValueOp uint8

KeyValueOp represents the type of KV operation (Put, Delete, Purge) Returned as part of watcher entry. 

const (
	KeyValuePut KeyValueOp = iota
	KeyValueDelete
	KeyValuePurge
)

func (KeyValueOp) String 

func (op KeyValueOp) String() string

type KeyValueStatus 

type KeyValueStatus interface {
	// Bucket the name of the bucket
	Bucket() string

	// Values is how many messages are in the bucket, including historical values
	Values() uint64

	// History returns the configured history kept per key
	History() int64

	// TTL is how long the bucket keeps values for
	TTL() time.Duration

	// BackingStore indicates what technology is used for storage of the bucket
	BackingStore() string

	// Bytes returns the size in bytes of the bucket
	Bytes() uint64

	// IsCompressed indicates if the data is compressed on disk
	IsCompressed() bool
}

KeyValueStatus is run-time status about a Key-Value bucket

type KeyWatcher 

type KeyWatcher interface {
	// Updates returns a channel to read any updates to entries.
	Updates() <-chan KeyValueEntry
	// Stop will stop this watcher.
	Stop() error
}

KeyWatcher is what is returned when doing a watch.

type ListObjectsOpt 

type ListObjectsOpt func(opts *listObjectOpts) error

func ListObjectsShowDeleted 

func ListObjectsShowDeleted() ListObjectsOpt

ListObjectsShowDeleted makes ListObjects() return deleted objects.

type MessageBatch 

type MessageBatch interface {
	Messages() <-chan Msg
	Error() error
}

type MessageHandler 

type MessageHandler func(msg Msg)

MessageHandler is a handler function used as callback in [Consume]

type MessagesContext 

type MessagesContext interface {
	// Next retreives next message on a stream. It will block until the next
	// message is available.
	Next() (Msg, error)

	// Stop unsubscribes from the stream and cancels subscription. Calling
	// Next after calling Stop will return ErrMsgIteratorClosed error.
	// All messages that are already in the buffer are discarded.
	Stop()

	// Drain unsubscribes from the stream and cancels subscription. All
	// messages that are already in the buffer will be available on
	// subsequent calls to Next. After the buffer is drained, Next will
	// return ErrMsgIteratorClosed error.
	Drain()
}

MessagesContext supports iterating over a messages on a stream.

type Msg 

type Msg interface {
	// Metadata returns [MsgMetadata] for a JetStream message
	Metadata() (*MsgMetadata, error)
	// Data returns the message body
	Data() []byte
	// Headers returns a map of headers for a message
	Headers() nats.Header
	// Subject returns a subject on which a message is published
	Subject() string
	// Reply returns a reply subject for a message
	Reply() string

	// Ack acknowledges a message
	// This tells the server that the message was successfully processed and it can move on to the next message
	Ack() error
	// DoubleAck acknowledges a message and waits for ack from server
	DoubleAck(context.Context) error
	// Nak negatively acknowledges a message
	// This tells the server to redeliver the message
	Nak() error
	// NakWithDelay negatively acknowledges a message
	// This tells the server to redeliver the message
	// after the given `delay` duration
	NakWithDelay(delay time.Duration) error
	// InProgress tells the server that this message is being worked on
	// It resets the redelivery timer on the server
	InProgress() error
	// Term tells the server to not redeliver this message, regardless of the value of nats.MaxDeliver
	Term() error
}

Msg contains methods to operate on a JetStream message Metadata, Data, Headers, Subject and Reply can be used to retrieve the specific parts of the underlying message Ack, DoubleAck, Nak, InProgress and Term are various flavors of ack requests

type MsgErrHandler 

type MsgErrHandler func(JetStream, *nats.Msg, error)

MsgErrHandler is used to process asynchronous errors from JetStream PublishAsync. It will return the original message sent to the server for possible retransmitting and the error encountered.

type MsgMetadata 

type MsgMetadata struct {
	Sequence     SequencePair
	NumDelivered uint64
	NumPending   uint64
	Timestamp    time.Time
	Stream       string
	Consumer     string
	Domain       string
}

MsgMetadata is the JetStream metadata associated with received messages.

type ObjectBucketStatus 

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

ObjectBucketStatus represents status of a Bucket, implements ObjectStoreStatus

func (*ObjectBucketStatus) BackingStore 

func (s *ObjectBucketStatus) BackingStore() string

BackingStore indicates what technology is used for storage of the bucket

func (*ObjectBucketStatus) Bucket 

func (s *ObjectBucketStatus) Bucket() string

Bucket is the name of the bucket

func (*ObjectBucketStatus) Description 

func (s *ObjectBucketStatus) Description() string

Description is the description supplied when creating the bucket

func (*ObjectBucketStatus) Metadata 

func (s *ObjectBucketStatus) Metadata() map[string]string

Metadata is the metadata supplied when creating the bucket

func (*ObjectBucketStatus) Replicas 

func (s *ObjectBucketStatus) Replicas() int

Replicas indicates how many storage replicas are kept for the data in the bucket

func (*ObjectBucketStatus) Sealed 

func (s *ObjectBucketStatus) Sealed() bool

Sealed indicates the stream is sealed and cannot be modified in any way

func (*ObjectBucketStatus) Size 

func (s *ObjectBucketStatus) Size() uint64

Size is the combined size of all data in the bucket including metadata, in bytes

func (*ObjectBucketStatus) Storage 

func (s *ObjectBucketStatus) Storage() StorageType

Storage indicates the underlying JetStream storage technology used to store data

func (*ObjectBucketStatus) StreamInfo 

func (s *ObjectBucketStatus) StreamInfo() *StreamInfo

StreamInfo is the stream info retrieved to create the status

func (*ObjectBucketStatus) TTL 

func (s *ObjectBucketStatus) TTL() time.Duration

TTL indicates how long objects are kept in the bucket

type ObjectInfo 

type ObjectInfo struct {
	ObjectMeta
	Bucket  string    `json:"bucket"`
	NUID    string    `json:"nuid"`
	Size    uint64    `json:"size"`
	ModTime time.Time `json:"mtime"`
	Chunks  uint32    `json:"chunks"`
	Digest  string    `json:"digest,omitempty"`
	Deleted bool      `json:"deleted,omitempty"`
}

ObjectInfo is meta plus instance information. 

type ObjectLink struct {
	// Bucket is the name of the other object store.
	Bucket string `json:"bucket"`
	// Name can be used to link to a single object.
	// If empty means this is a link to the whole store, like a directory.
	Name string `json:"name,omitempty"`
}

ObjectLink is used to embed links to other buckets and objects.

type ObjectMeta 

type ObjectMeta struct {
	Name        string            `json:"name"`
	Description string            `json:"description,omitempty"`
	Headers     nats.Header       `json:"headers,omitempty"`
	Metadata    map[string]string `json:"metadata,omitempty"`

	// Additional options.
	Opts *ObjectMetaOptions `json:"options,omitempty"`
}

ObjectMeta is high level information about an object.

type ObjectMetaOptions 

type ObjectMetaOptions struct {
	Link      *ObjectLink `json:"link,omitempty"`
	ChunkSize uint32      `json:"max_chunk_size,omitempty"`
}

ObjectMetaOptions

type ObjectResult 

type ObjectResult interface {
	io.ReadCloser
	Info() (*ObjectInfo, error)
	Error() error
}

ObjectResult will return the underlying stream info and also be an io.ReadCloser.

type ObjectStore 

type ObjectStore interface {
	// Put will place the contents from the reader into a new object.
	Put(ctx context.Context, obj ObjectMeta, reader io.Reader) (*ObjectInfo, error)
	// Get will pull the named object from the object store.
	Get(ctx context.Context, name string, opts ...GetObjectOpt) (ObjectResult, error)

	// PutBytes is convenience function to put a byte slice into this object store.
	PutBytes(ctx context.Context, name string, data []byte) (*ObjectInfo, error)
	// GetBytes is a convenience function to pull an object from this object store and return it as a byte slice.
	GetBytes(ctx context.Context, name string, opts ...GetObjectOpt) ([]byte, error)

	// PutString is convenience function to put a string into this object store.
	PutString(ctx context.Context, name string, data string) (*ObjectInfo, error)
	// GetString is a convenience function to pull an object from this object store and return it as a string.
	GetString(ctx context.Context, name string, opts ...GetObjectOpt) (string, error)

	// PutFile is convenience function to put a file into this object store.
	PutFile(ctx context.Context, file string) (*ObjectInfo, error)
	// GetFile is a convenience function to pull an object from this object store and place it in a file.
	GetFile(ctx context.Context, name, file string, opts ...GetObjectOpt) error

	// GetInfo will retrieve the current information for the object.
	GetInfo(ctx context.Context, name string, opts ...GetObjectInfoOpt) (*ObjectInfo, error)
	// UpdateMeta will update the metadata for the object.
	UpdateMeta(ctx context.Context, name string, meta ObjectMeta) error

	// Delete will delete the named object.
	Delete(ctx context.Context, name string) error

	// AddLink will add a link to another object.
	AddLink(ctx context.Context, name string, obj *ObjectInfo) (*ObjectInfo, error)

	// AddBucketLink will add a link to another object store.
	AddBucketLink(ctx context.Context, name string, bucket ObjectStore) (*ObjectInfo, error)

	// Seal will seal the object store, no further modifications will be allowed.
	Seal(ctx context.Context) error

	// Watch for changes in the underlying store and receive meta information updates.
	Watch(ctx context.Context, opts ...WatchOpt) (ObjectWatcher, error)

	// List will list all the objects in this store.
	List(ctx context.Context, opts ...ListObjectsOpt) ([]*ObjectInfo, error)

	// Status retrieves run-time status about the backing store of the bucket.
	Status(ctx context.Context) (ObjectStoreStatus, error)
}

ObjectStore is a blob store capable of storing large objects efficiently in JetStream streams

type ObjectStoreConfig 

type ObjectStoreConfig struct {
	Bucket      string        `json:"bucket"`
	Description string        `json:"description,omitempty"`
	TTL         time.Duration `json:"max_age,omitempty"`
	MaxBytes    int64         `json:"max_bytes,omitempty"`
	Storage     StorageType   `json:"storage,omitempty"`
	Replicas    int           `json:"num_replicas,omitempty"`
	Placement   *Placement    `json:"placement,omitempty"`

	// Bucket-specific metadata
	// NOTE: Metadata requires nats-server v2.10.0+
	Metadata map[string]string `json:"metadata,omitempty"`
}

ObjectStoreConfig is the config for the object store.

type ObjectStoreManager 

type ObjectStoreManager interface {
	// ObjectStore will look up and bind to an existing object store instance.
	ObjectStore(ctx context.Context, bucket string) (ObjectStore, error)
	// CreateObjectStore will create an object store.
	CreateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error)
	// DeleteObjectStore will delete the underlying stream for the named object.
	DeleteObjectStore(ctx context.Context, bucket string) error
	// ObjectStoreNames is used to retrieve a list of bucket names
	ObjectStoreNames(ctx context.Context) ObjectStoreNamesLister
	// ObjectStores is used to retrieve a list of bucket statuses
	ObjectStores(ctx context.Context) ObjectStoresLister
}

ObjectStoreManager creates, loads and deletes Object Stores

type ObjectStoreNamesLister 

type ObjectStoreNamesLister interface {
	Name() <-chan string
	Error() error
}

type ObjectStoreStatus 

type ObjectStoreStatus interface {
	// Bucket is the name of the bucket
	Bucket() string
	// Description is the description supplied when creating the bucket
	Description() string
	// TTL indicates how long objects are kept in the bucket
	TTL() time.Duration
	// Storage indicates the underlying JetStream storage technology used to store data
	Storage() StorageType
	// Replicas indicates how many storage replicas are kept for the data in the bucket
	Replicas() int
	// Sealed indicates the stream is sealed and cannot be modified in any way
	Sealed() bool
	// Size is the combined size of all data in the bucket including metadata, in bytes
	Size() uint64
	// BackingStore provides details about the underlying storage
	BackingStore() string
	// Metadata is the user supplied metadata for the bucket
	Metadata() map[string]string
}

type ObjectStoresLister 

type ObjectStoresLister interface {
	Status() <-chan ObjectStoreStatus
	Error() error
}

type ObjectWatcher 

type ObjectWatcher interface {
	// Updates returns a channel to read any updates to entries.
	Updates() <-chan *ObjectInfo
	// Stop will stop this watcher.
	Stop() error
}

ObjectWatcher is what is returned when doing a watch.

type OrderedConsumerConfig 

type OrderedConsumerConfig struct {
	FilterSubjects    []string      `json:"filter_subjects,omitempty"`
	DeliverPolicy     DeliverPolicy `json:"deliver_policy"`
	OptStartSeq       uint64        `json:"opt_start_seq,omitempty"`
	OptStartTime      *time.Time    `json:"opt_start_time,omitempty"`
	ReplayPolicy      ReplayPolicy  `json:"replay_policy"`
	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
	HeadersOnly       bool          `json:"headers_only,omitempty"`

	// Maximum number of attempts for the consumer to be recreated
	// Defaults to unlimited
	MaxResetAttempts int
}

type PeerInfo 

type PeerInfo struct {
	Name    string        `json:"name"`
	Current bool          `json:"current"`
	Offline bool          `json:"offline,omitempty"`
	Active  time.Duration `json:"active"`
	Lag     uint64        `json:"lag,omitempty"`
}

PeerInfo shows information about all the peers in the cluster that are supporting the stream or consumer.

type Placement 

type Placement struct {
	Cluster string   `json:"cluster"`
	Tags    []string `json:"tags,omitempty"`
}

Placement is used to guide placement of streams in clustered JetStream.

type PubAck 

type PubAck struct {
	Stream    string `json:"stream"`
	Sequence  uint64 `json:"seq"`
	Duplicate bool   `json:"duplicate,omitempty"`
	Domain    string `json:"domain,omitempty"`
}

PubAck is an ack received after successfully publishing a message.

type PubAckFuture 

type PubAckFuture interface {
	// Ok returns a receive only channel that can be used to get a PubAck.
	Ok() <-chan *PubAck

	// Err returns a receive only channel that can be used to get the error from an async publish.
	Err() <-chan error

	// Msg returns the message that was sent to the server.
	Msg() *nats.Msg
}

PubAckFuture is a future for a PubAck.

type PublishOpt 

type PublishOpt func(*pubOpts) error

func WithExpectLastMsgID 

func WithExpectLastMsgID(id string) PublishOpt

WithExpectLastMsgID sets the expected last msgId in the response from the publish.

func WithExpectLastSequence 

func WithExpectLastSequence(seq uint64) PublishOpt

WithExpectLastSequence sets the expected sequence in the response from the publish.

func WithExpectLastSequencePerSubject 

func WithExpectLastSequencePerSubject(seq uint64) PublishOpt

WithExpectLastSequencePerSubject sets the expected sequence per subject in the response from the publish.

func WithExpectStream 

func WithExpectStream(stream string) PublishOpt

WithExpectStream sets the expected stream to respond from the publish.

func WithMsgID 

func WithMsgID(id string) PublishOpt

WithMsgID sets the message ID used for deduplication.

func WithRetryAttempts 

func WithRetryAttempts(num int) PublishOpt

WithRetryAttempts sets the retry number of attempts when ErrNoResponders is encountered. Defaults to 2

func WithRetryWait 

func WithRetryWait(dur time.Duration) PublishOpt

WithRetryWait sets the retry wait time when ErrNoResponders is encountered. Defaults to 250ms

func WithStallWait 

func WithStallWait(ttl time.Duration) PublishOpt

WithStallWait sets the max wait when the producer becomes stall producing messages.

type Publisher 

type Publisher interface {
	// Publish performs a synchronous publish to a stream and waits for ack from server
	// It accepts subject name (which must be bound to a stream) and message data
	Publish(ctx context.Context, subject string, payload []byte, opts ...PublishOpt) (*PubAck, error)
	// PublishMsg performs a synchronous publish to a stream and waits for ack from server
	// It accepts subject name (which must be bound to a stream) and nats.Message
	PublishMsg(ctx context.Context, msg *nats.Msg, opts ...PublishOpt) (*PubAck, error)
	// PublishAsync performs an asynchronous publish to a stream and returns [PubAckFuture] interface
	// It accepts subject name (which must be bound to a stream) and message data
	PublishAsync(subject string, payload []byte, opts ...PublishOpt) (PubAckFuture, error)
	// PublishMsgAsync performs an asynchronous publish to a stream and returns [PubAckFuture] interface
	// It accepts subject name (which must be bound to a stream) and nats.Message
	PublishMsgAsync(msg *nats.Msg, opts ...PublishOpt) (PubAckFuture, error)
	// PublishAsyncPending returns the number of async publishes outstanding for this context
	PublishAsyncPending() int
	// PublishAsyncComplete returns a channel that will be closed when all outstanding messages are ack'd
	PublishAsyncComplete() <-chan struct{}
}

type PullConsumeOpt 

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

PullConsumeOpt represent additional options used in [Consume] for pull consumers

func ConsumeErrHandler 

func ConsumeErrHandler(cb ConsumeErrHandlerFunc) PullConsumeOpt

ConsumeErrHandler sets custom error handler invoked when an error was encountered while consuming messages It will be invoked for both terminal (Consumer Deleted, invalid request body) and non-terminal (e.g. missing heartbeats) errors

type PullExpiry 

type PullExpiry time.Duration

PullExpiry sets timeout on a single batch request, waiting until at least one message is available

type PullHeartbeat 

type PullHeartbeat time.Duration

PullHeartbeat sets the idle heartbeat duration for a pull subscription If a client does not receive a heartbeat message from a stream for more than the idle heartbeat setting, the subscription will be removed and error will be passed to the message handler

type PullMaxBytes 

type PullMaxBytes int

PullMaxBytes sets max_bytes limit on a fetch request

type PullMaxMessages 

type PullMaxMessages int

PullMaxMessages limits the number of messages to be fetched from the stream in one request If not provided, a default of 100 messages will be used

type PullMessagesOpt 

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

PullMessagesOpt represent additional options used in [Messages] for pull consumers

func WithMessagesErrOnMissingHeartbeat 

func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt

WithMessagesErrOnMissingHeartbeat sets whether a missing heartbeat error should be reported when calling Next (Default: true).

type PullThresholdBytes 

type PullThresholdBytes int

PullThresholdBytes sets the byte count on which Consume will trigger new pull request to the server. Defaults to 50% of MaxBytes (if set).

type PullThresholdMessages 

type PullThresholdMessages int

PullThresholdMessages sets the message count on which Consume will trigger new pull request to the server. Defaults to 50% of MaxMessages.

type RawStreamMsg 

type RawStreamMsg struct {
	Subject  string
	Sequence uint64
	Header   nats.Header
	Data     []byte
	Time     time.Time
}

type RePublish 

type RePublish struct {
	Source      string `json:"src,omitempty"`
	Destination string `json:"dest"`
	HeadersOnly bool   `json:"headers_only,omitempty"`
}

RePublish is for republishing messages once committed to a stream. The original subject is remapped from the subject pattern to the destination pattern.

type ReplayPolicy 

type ReplayPolicy int

ReplayPolicy determines how the consumer should replay messages it already has queued in the stream. 

const (
	// ReplayInstantPolicy will replay messages as fast as possible.
	ReplayInstantPolicy ReplayPolicy = iota

	// ReplayOriginalPolicy will maintain the same timing as the messages were received.
	ReplayOriginalPolicy
)

func (ReplayPolicy) MarshalJSON 

func (p ReplayPolicy) MarshalJSON() ([]byte, error)

func (ReplayPolicy) String 

func (p ReplayPolicy) String() string

func (*ReplayPolicy) UnmarshalJSON 

func (p *ReplayPolicy) UnmarshalJSON(data []byte) error

type RetentionPolicy 

type RetentionPolicy int

RetentionPolicy determines how messages in a set are retained. 

const (
	// LimitsPolicy (default) means that messages are retained until any given limit is reached.
	// This could be one of MaxMsgs, MaxBytes, or MaxAge.
	LimitsPolicy RetentionPolicy = iota
	// InterestPolicy specifies that when all known observables have acknowledged a message it can be removed.
	InterestPolicy
	// WorkQueuePolicy specifies that when the first worker or subscriber acknowledges the message it can be removed.
	WorkQueuePolicy
)

func (RetentionPolicy) MarshalJSON 

func (rp RetentionPolicy) MarshalJSON() ([]byte, error)

func (RetentionPolicy) String 

func (rp RetentionPolicy) String() string

func (*RetentionPolicy) UnmarshalJSON 

func (rp *RetentionPolicy) UnmarshalJSON(data []byte) error

type SequenceInfo 

type SequenceInfo struct {
	Consumer uint64     `json:"consumer_seq"`
	Stream   uint64     `json:"stream_seq"`
	Last     *time.Time `json:"last_active,omitempty"`
}

SequenceInfo has both the consumer and the stream sequence and last activity.

type SequencePair 

type SequencePair struct {
	Consumer uint64 `json:"consumer_seq"`
	Stream   uint64 `json:"stream_seq"`
}

SequencePair includes the consumer and stream sequence info from a JetStream consumer.

type StopAfter 

type StopAfter int

StopAfter sets the number of messages after which the consumer is automatically stopped

type StorageType 

type StorageType int

StorageType determines how messages are stored for retention. 

const (
	// FileStorage specifies on disk storage. It's the default.
	FileStorage StorageType = iota
	// MemoryStorage specifies in memory only.
	MemoryStorage
)

func (StorageType) MarshalJSON 

func (st StorageType) MarshalJSON() ([]byte, error)

func (StorageType) String 

func (st StorageType) String() string

func (*StorageType) UnmarshalJSON 

func (st *StorageType) UnmarshalJSON(data []byte) error

type StoreCompression 

type StoreCompression uint8

StoreCompression determines how messages are compressed. 

const (
	NoCompression StoreCompression = iota
	S2Compression
)

func (StoreCompression) MarshalJSON 

func (alg StoreCompression) MarshalJSON() ([]byte, error)

func (StoreCompression) String 

func (alg StoreCompression) String() string

func (*StoreCompression) UnmarshalJSON 

func (alg *StoreCompression) UnmarshalJSON(b []byte) error

type Stream 

type Stream interface {

	// Info returns stream details
	Info(ctx context.Context, opts ...StreamInfoOpt) (*StreamInfo, error)
	// CachedInfo returns *StreamInfo cached on a consumer struct
	CachedInfo() *StreamInfo

	// Purge removes messages from a stream
	Purge(ctx context.Context, opts ...StreamPurgeOpt) error

	// GetMsg retrieves a raw stream message stored in JetStream by sequence number
	GetMsg(ctx context.Context, seq uint64, opts ...GetMsgOpt) (*RawStreamMsg, error)
	// GetLastMsgForSubject retrieves the last raw stream message stored in JetStream by subject
	GetLastMsgForSubject(ctx context.Context, subject string) (*RawStreamMsg, error)
	// DeleteMsg deletes a message from a stream.
	// The message is marked as erased, but not overwritten
	DeleteMsg(ctx context.Context, seq uint64) error
	// SecureDeleteMsg deletes a message from a stream. The deleted message is overwritten with random data
	// As a result, this operation is slower than DeleteMsg()
	SecureDeleteMsg(ctx context.Context, seq uint64) error
	// contains filtered or unexported methods
}

Stream contains CRUD methods on a consumer, as well as operations on an existing stream

type StreamConfig 

type StreamConfig struct {
	Name                 string           `json:"name"`
	Description          string           `json:"description,omitempty"`
	Subjects             []string         `json:"subjects,omitempty"`
	Retention            RetentionPolicy  `json:"retention"`
	MaxConsumers         int              `json:"max_consumers"`
	MaxMsgs              int64            `json:"max_msgs"`
	MaxBytes             int64            `json:"max_bytes"`
	Discard              DiscardPolicy    `json:"discard"`
	DiscardNewPerSubject bool             `json:"discard_new_per_subject,omitempty"`
	MaxAge               time.Duration    `json:"max_age"`
	MaxMsgsPerSubject    int64            `json:"max_msgs_per_subject"`
	MaxMsgSize           int32            `json:"max_msg_size,omitempty"`
	Storage              StorageType      `json:"storage"`
	Replicas             int              `json:"num_replicas"`
	NoAck                bool             `json:"no_ack,omitempty"`
	Template             string           `json:"template_owner,omitempty"`
	Duplicates           time.Duration    `json:"duplicate_window,omitempty"`
	Placement            *Placement       `json:"placement,omitempty"`
	Mirror               *StreamSource    `json:"mirror,omitempty"`
	Sources              []*StreamSource  `json:"sources,omitempty"`
	Sealed               bool             `json:"sealed,omitempty"`
	DenyDelete           bool             `json:"deny_delete,omitempty"`
	DenyPurge            bool             `json:"deny_purge,omitempty"`
	AllowRollup          bool             `json:"allow_rollup_hdrs,omitempty"`
	Compression          StoreCompression `json:"compression"`
	FirstSeq             uint64           `json:"first_seq,omitempty"`

	// Allow applying a subject transform to incoming messages before doing anything else
	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`

	// Allow republish of the message after being sequenced and stored.
	RePublish *RePublish `json:"republish,omitempty"`

	// Allow higher performance, direct access to get individual messages. E.g. KeyValue
	AllowDirect bool `json:"allow_direct"`
	// Allow higher performance and unified direct access for mirrors as well.
	MirrorDirect bool `json:"mirror_direct"`

	// Limits for consumers on this stream.
	ConsumerLimits StreamConsumerLimits `json:"consumer_limits,omitempty"`

	// Metadata is additional metadata for the Stream.
	// Keys starting with `_nats` are reserved.
	// NOTE: Metadata requires nats-server v2.10.0+
	Metadata map[string]string `json:"metadata,omitempty"`
}

type StreamConsumerLimits 

type StreamConsumerLimits struct {
	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
	MaxAckPending     int           `json:"max_ack_pending,omitempty"`
}

StreamConsumerLimits are the limits for a consumer on a stream. These can be overridden on a per consumer basis.

type StreamConsumerManager 

type StreamConsumerManager interface {
	// CreateOrUpdateConsumer creates a consumer on a given stream with given config.
	// If consumer already exists, it will be updated (if possible).
	// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
	CreateOrUpdateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
	// CreateConsumer creates a consumer on a given stream with given config.
	// If consumer already exists, ErrConsumerExists is returned.
	// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
	CreateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
	// UpdateConsumer updates an existing consumer.
	// If consumer does not exist, ErrConsumerDoesNotExist is returned.
	// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
	UpdateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
	// OrderedConsumer returns an OrderedConsumer instance.
	// OrderedConsumer allows fetching messages from a stream (just like standard consumer),
	// for in order delivery of messages. Underlying consumer is re-created when necessary,
	// without additional client code.
	OrderedConsumer(ctx context.Context, stream string, cfg OrderedConsumerConfig) (Consumer, error)
	// Consumer returns a hook to an existing consumer, allowing processing of messages
	Consumer(ctx context.Context, stream string, consumer string) (Consumer, error)
	// DeleteConsumer removes a consumer with given name from a stream
	DeleteConsumer(ctx context.Context, stream string, consumer string) error
}

type StreamInfo 

type StreamInfo struct {
	Config  StreamConfig        `json:"config"`
	Created time.Time           `json:"created"`
	State   StreamState         `json:"state"`
	Cluster *ClusterInfo        `json:"cluster,omitempty"`
	Mirror  *StreamSourceInfo   `json:"mirror,omitempty"`
	Sources []*StreamSourceInfo `json:"sources,omitempty"`
}

StreamInfo shows config and current state for this stream.

type StreamInfoLister 

type StreamInfoLister interface {
	Info() <-chan *StreamInfo
	Err() error
}

type StreamInfoOpt 

type StreamInfoOpt func(*streamInfoRequest) error

func WithDeletedDetails 

func WithDeletedDetails(deletedDetails bool) StreamInfoOpt

WithDeletedDetails can be used to display the information about messages deleted from a stream on a stream info request

func WithSubjectFilter 

func WithSubjectFilter(subject string) StreamInfoOpt

WithSubjectFilter can be used to display the information about messages stored on given subjects. NOTE: if the subject filter matches over 100k subjects, this will result in multiple requests to the server to retrieve all the information, and all of the returned subjects will be kept in memory.

type StreamListOpt 

type StreamListOpt func(*streamsRequest) error

func WithStreamListSubject 

func WithStreamListSubject(subject string) StreamListOpt

WithStreamListSubject can be used to filter results of ListStreams and StreamNames requests to only streams that have given subject in their configuration

type StreamManager 

type StreamManager interface {
	// CreateStream creates a new stream with given config and returns a hook to operate on it
	CreateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
	// UpdateStream updates an existing stream
	UpdateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
	// CreateOrUpdateStream creates a stream with given config. If stream already exists, it will be updated (if possible).
	CreateOrUpdateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
	// Stream returns a [Stream] hook for a given stream name
	Stream(ctx context.Context, stream string) (Stream, error)
	// StreamNameBySubject returns a stream name stream listening on given subject
	StreamNameBySubject(ctx context.Context, subject string) (string, error)
	// DeleteStream removes a stream with given name
	DeleteStream(ctx context.Context, stream string) error
	// ListStreams returns StreamInfoLister enabling iterating over a channel of stream infos
	ListStreams(context.Context, ...StreamListOpt) StreamInfoLister
	// StreamNames returns a  StreamNameLister enabling iterating over a channel of stream names
	StreamNames(context.Context, ...StreamListOpt) StreamNameLister
}

type StreamNameLister 

type StreamNameLister interface {
	Name() <-chan string
	Err() error
}

type StreamPurgeOpt 

type StreamPurgeOpt func(*StreamPurgeRequest) error

func WithPurgeKeep 

func WithPurgeKeep(keep uint64) StreamPurgeOpt

WithPurgeKeep sets the number of messages to be kept in the stream after purge. Can be combined with WithPurgeSubject option, but not with WithPurgeSequence

func WithPurgeSequence 

func WithPurgeSequence(sequence uint64) StreamPurgeOpt

WithPurgeSequence is used to set a specific sequence number up to which (but not including) messages will be purged from a stream Can be combined with WithPurgeSubject option, but not with WithPurgeKeep

func WithPurgeSubject 

func WithPurgeSubject(subject string) StreamPurgeOpt

WithPurgeSubject sets a specific subject for which messages on a stream will be purged

type StreamPurgeRequest 

type StreamPurgeRequest struct {
	// Purge up to but not including sequence.
	Sequence uint64 `json:"seq,omitempty"`
	// Subject to match against messages for the purge command.
	Subject string `json:"filter,omitempty"`
	// Number of messages to keep.
	Keep uint64 `json:"keep,omitempty"`
}

type StreamSource 

type StreamSource struct {
	Name              string                   `json:"name"`
	OptStartSeq       uint64                   `json:"opt_start_seq,omitempty"`
	OptStartTime      *time.Time               `json:"opt_start_time,omitempty"`
	FilterSubject     string                   `json:"filter_subject,omitempty"`
	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
	External          *ExternalStream          `json:"external,omitempty"`
	Domain            string                   `json:"-"`
}

StreamSource dictates how streams can source from other streams.

type StreamSourceInfo 

type StreamSourceInfo struct {
	Name              string                   `json:"name"`
	Lag               uint64                   `json:"lag"`
	Active            time.Duration            `json:"active"`
	FilterSubject     string                   `json:"filter_subject,omitempty"`
	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
}

StreamSourceInfo shows information about an upstream stream source.

type StreamState 

type StreamState struct {
	Msgs        uint64            `json:"messages"`
	Bytes       uint64            `json:"bytes"`
	FirstSeq    uint64            `json:"first_seq"`
	FirstTime   time.Time         `json:"first_ts"`
	LastSeq     uint64            `json:"last_seq"`
	LastTime    time.Time         `json:"last_ts"`
	Consumers   int               `json:"consumer_count"`
	Deleted     []uint64          `json:"deleted"`
	NumDeleted  int               `json:"num_deleted"`
	NumSubjects uint64            `json:"num_subjects"`
	Subjects    map[string]uint64 `json:"subjects"`
}

StreamState is information about the given stream.

type SubjectTransformConfig 

type SubjectTransformConfig struct {
	Source      string `json:"src"`
	Destination string `json:"dest"`
}

SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received

type Tier 

type Tier struct {
	Memory         uint64        `json:"memory"`
	Store          uint64        `json:"storage"`
	ReservedMemory uint64        `json:"reserved_memory"`
	ReservedStore  uint64        `json:"reserved_storage"`
	Streams        int           `json:"streams"`
	Consumers      int           `json:"consumers"`
	Limits         AccountLimits `json:"limits"`
}

type WatchOpt 

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

func IgnoreDeletes 

func IgnoreDeletes() WatchOpt

IgnoreDeletes will have the key watcher not pass any deleted keys.

func IncludeHistory 

func IncludeHistory() WatchOpt

IncludeHistory instructs the key watcher to include historical values as well.

func MetaOnly 

func MetaOnly() WatchOpt

MetaOnly instructs the key watcher to retrieve only the entry meta data, not the entry value.

func ResumeFromRevision 

func ResumeFromRevision(revision uint64) WatchOpt

ResumeFromRevision instructs the key watcher to resume from a specific revision number.

func UpdatesOnly 

func UpdatesOnly() WatchOpt

UpdatesOnly instructs the key watcher to only include updates on values (without latest values when started).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.