nats

package module

Go to main page

Versions in this module

v0

v0.7.0

Aug 22, 2026

Changes in this version

+ const ApplicationJSON

+ const ApplicationMsgPack

+ const ApplicationProtobuf

+ const ApplicationVNDGoogleProtobuf

+ const ApplicationXMsgPack

+ const ApplicationXProtobuf

+ const HeaderContractEncoding

+ const HeaderHTTPContentType

+ const HeaderNATSContentType

+ const HeaderNatsContentType

+ const HeaderSchema

+ const HeaderSchemaVersion

+ var ErrSchemaMismatch = errors.New("schema contract mismatch")

+ func ApplySchemaHeaders(msg *Message, name, version, encoding string)

+ func CheckSchema(msg *natspkg.Msg, expect SchemaExpectation) error

+ func DecodeMsgWithSchema(msg *natspkg.Msg, typ MessageType, dst any, expect SchemaExpectation) error

+ func SchemaFromHeader(h natspkg.Header) (string, string, string)

+ type CircuitBreaker struct

+ func NewCircuitBreaker(cfg CircuitConfig) *CircuitBreaker

+ func (b *CircuitBreaker) IsOpen() bool

+ func (b *CircuitBreaker) Opens() int64

+ func (b *CircuitBreaker) PauseDuration() time.Duration

+ func (b *CircuitBreaker) ShouldPause(d AdaptivePressureDecision) bool

+ type CircuitConfig struct

+ Enabled bool

+ OpenOnLevel AdaptivePressureLevel

+ PauseDuration time.Duration

type Client

+ OrderedSubscribe func(ctx context.Context, stream string, cfg OrderedConsumerConfig, ...) (OrderedSubscription, error)

type Config

+ Circuit CircuitConfig

type Connection

+ Domain string

type Consumer

+ OrderedSubscribe func(ctx context.Context, stream string, cfg OrderedConsumerConfig, ...) (OrderedSubscription, error)

type IncidentTrigger

+ func (it IncidentTrigger) String() string

type Incidents

+ AttachToHandler func(stream, durable string, cfg CapsuleAutoConfig, handler MsgHandler) MsgHandler

type MsgHandler

+ func AttachIncidentsHandler(inc Incidents, stream, durable string, cfg CapsuleAutoConfig, ...) MsgHandler

+ type OrderedConsumerConfig struct

+ DeliverPolicy DeliverPolicy

+ FilterSubject string

+ FilterSubjects []string

+ HasDeliverPolicy bool

+ HasReplayPolicy bool

+ HeadersOnly bool

+ InactiveThreshold time.Duration

+ NamePrefix string

+ OptStartSeq uint64

+ OptStartTime *time.Time

+ ReplayPolicy ReplayPolicy

+ type OrderedSubscription interface

+ Stop func()

+ Stopped func() <-chan struct{}

+ type SchedulePublishConfig struct

+ SubjectTemplate string

+ type SchemaExpectation struct

+ Encoding string

+ Name string

+ Required bool

+ Version string

v0.6.0

Aug 2, 2026

Changes in this version

+ const DefaultIncidentCapsuleBucket

+ const DefaultIncidentIndexBucket

+ const EncodingBrotli

+ const EncodingGzip

+ const HeaderContentEncoding

+ const MaxPayloadDecompressBytes

+ const MinPayloadCompressBytes

+ var ErrPayloadTooLarge = errors.New("decompressed payload exceeds size limit")

+ var ErrUnsupportedContentEncoding = errors.New("unsupported content encoding")

+ func DecodeMsgWithDecompress(msg *natspkg.Msg, typ MessageType, dst any) error

+ func DecompressPayload(data []byte, encoding string) ([]byte, error)

+ func MaybeCompressPayload(plain []byte) ([]byte, string, bool)

+ type AdaptivePressure struct

+ func NewAdaptivePressure(cfg AdaptivePressureConfig) *AdaptivePressure

+ func (a *AdaptivePressure) Decision() AdaptivePressureDecision

+ func (a *AdaptivePressure) FetchBatchOr(defaultBatch int) int

+ func (a *AdaptivePressure) Observe(in AdaptivePressureInput) AdaptivePressureDecision

+ type AdaptivePressureConfig struct

+ Enabled bool

+ MaxFetchBatch int

+ MinAckPending int

+ MinFetchBatch int

+ NakDelayMax time.Duration

+ NakDelayMin time.Duration

+ type AdaptivePressureDecision struct

+ AckPending int

+ FetchBatch int

+ Level AdaptivePressureLevel

+ NakDelay time.Duration

+ func DecideAdaptivePressure(cfg AdaptivePressureConfig, in AdaptivePressureInput) AdaptivePressureDecision

+ type AdaptivePressureInput struct

+ BaselineLatency time.Duration

+ CurrentAckPend int

+ CurrentBatch int

+ HandlerLatency time.Duration

+ Lag uint64

+ MaxAckPending int

+ Pending uint64

+ PoolCapacity int

+ PoolDepth int

+ type AdaptivePressureLevel uint8

+ const PressureCritical

+ const PressureElevated

+ const PressureHard

+ const PressureIdle

+ const PressureWarm

+ type Capsule struct

+ Consumer string

+ CreatedAt time.Time

+ FailingSeq uint64

+ Fingerprint *BehaviorFingerprintKVPayload

+ FlightTimeline []IncidentEvent

+ ID string

+ Messages []CapsuleMessage

+ Reason string

+ SchemaVersion int

+ Stream string

+ Subject string

+ Trigger IncidentTrigger

+ type CapsuleAuto struct

+ func NewCapsuleAuto(client Client, stream, durable string, cfg CapsuleAutoConfig) *CapsuleAuto

+ func (a *CapsuleAuto) DLQRecorder() dlq.Recorder

+ func (a *CapsuleAuto) OnAnomaly() func(BehaviorAnomalyEvent)

+ func (a *CapsuleAuto) ShadowRecorder() shadow.Recorder

+ type CapsuleAutoConfig struct

+ Enabled bool

+ FlightRecorder *FlightRecorder

+ IndexBucket string

+ OnReady func(*Capsule)

+ Redact func(*CapsuleMessage)

+ StoreBucket string

+ Window int

+ type CapsuleMessage struct

+ Data []byte

+ Header map[string][]string

+ Sequence uint64

+ Subject string

+ Time time.Time

type Client

+ Incidents func() Incidents

type Config

+ AdaptivePressure AdaptivePressureConfig

type DurableConsumerConfig

+ DeliverGroup string

+ DeliverSubject string

+ type IncidentCapture struct

+ Consumer string

+ FailingSeq uint64

+ Fingerprint *BehaviorFingerprintKVPayload

+ FlightRecorder *FlightRecorder

+ IndexBucket string

+ Reason string

+ Redact func(*CapsuleMessage)

+ StoreBucket string

+ Stream string

+ Subject string

+ Trigger IncidentTrigger

+ Window int

+ type IncidentTrigger string

+ const TriggerAnomaly

+ const TriggerDLQ

+ const TriggerManual

+ const TriggerShadowMismatch

+ const TriggerStall

+ type Incidents interface

+ Capture func(ctx context.Context, cfg IncidentCapture) (*Capsule, error)

+ List func(ctx context.Context, stream, consumer, indexBucket string) ([]string, error)

+ Load func(ctx context.Context, bucket, id string) (*Capsule, error)

+ ReplayLocal func(ctx context.Context, capsule *Capsule, handler MsgHandler) error

+ type PayloadCompressionMode uint8

+ const PayloadCompressionAuto

+ const PayloadCompressionBrotli

+ const PayloadCompressionGzip

+ const PayloadCompressionOff

type PublisherConfig

+ PayloadCompression PayloadCompressionMode

type RequesterConfig

+ PayloadCompression PayloadCompressionMode

+ PayloadDecompression bool

type Responder

+ RespondBytes func(msg *natspkg.Msg, data []byte) error

+ RespondJSON func(msg *natspkg.Msg, v any) error

+ RespondMsgPack func(msg *natspkg.Msg, v any) error

+ RespondProto func(msg *natspkg.Msg, v proto.Message) error

type ResponderConfig

+ PayloadCompression PayloadCompressionMode

+ PayloadDecompression bool

type RuntimeConsumerConfig

+ PayloadDecompression bool

v0.5.0

Aug 2, 2026

Changes in this version

+ var ErrMonitoringBodyTooLarge = errors.New("monitoring response body too large")

v0.4.0

Aug 1, 2026

Changes in this version

type PullConsumer

+ Close func() error

v0.3.0

Jul 31, 2026

Changes in this version

+ const DefaultBehaviorFingerprintKVBucket

+ const DefaultMsgRangeMax

+ const MetaReplayLimit

+ const MetaReplayUntilSeq

+ const SlowReasonAckPending

+ const SlowReasonLag

+ const SlowReasonPending

+ var ErrInvalidReplayBound = errors.New("invalid replay bound")

+ func BehaviorFingerprintKVKey(stream, durable string) string

+ func ConsumerLagMessages(streamLastSeq, deliveredStreamSeq uint64) uint64

+ func EvaluateBehaviorFingerprint(current, baseline BehaviorSnapshot, cfg BehaviorFingerprintConfig) bool

+ func EvaluateSlowConsumer(pending, lag uint64, ackPending, maxAckPending int, cfg SlowConsumerConfig) (bool, []string)

+ func MarshalBehaviorFingerprintKV(stream, durable string, anomaly bool, normal, current BehaviorSnapshot, ...) ([]byte, error)

+ func ReportBehaviorFingerprintKV(ctx context.Context, keys KeyValueKeys, bucket, stream, durable string, ...) error

+ type BehaviorAnomalyEvent struct

+ Current BehaviorSnapshot

+ Durable string

+ Normal BehaviorSnapshot

+ Stream string

+ SustainedFor time.Duration

+ type BehaviorFingerprint struct

+ func WatchBehaviorFingerprint(ctx context.Context, sub Subscription, cfg BehaviorFingerprintConfig, ...) (*BehaviorFingerprint, error)

+ func (b *BehaviorFingerprint) Anomalous() bool

+ func (b *BehaviorFingerprint) Events() <-chan BehaviorAnomalyEvent

+ func (b *BehaviorFingerprint) Observe(elapsed time.Duration)

+ func (b *BehaviorFingerprint) Snapshot() (BehaviorSnapshot, BehaviorSnapshot, bool)

+ func (b *BehaviorFingerprint) Stop()

+ type BehaviorFingerprintConfig struct

+ CircuitStop bool

+ LatencyFactor float64

+ MinSamples int

+ OnAnomaly func(BehaviorAnomalyEvent)

+ PollInterval time.Duration

+ RateTolerance float64

+ ReportBucket string

+ ReportKV KeyValueKeys

+ SustainFor time.Duration

+ Warmup time.Duration

+ Window time.Duration

+ type BehaviorFingerprintKVPayload struct

+ Anomaly bool

+ Current BehaviorFingerprintKVSnapshot

+ Durable string

+ Normal BehaviorFingerprintKVSnapshot

+ Stream string

+ SustainedForMs int64

+ UpdatedAt time.Time

+ type BehaviorFingerprintKVSnapshot struct

+ MsgPerMin float64

+ ProcessingMs float64

+ func SnapshotToKV(s BehaviorSnapshot) BehaviorFingerprintKVSnapshot

+ type BehaviorSnapshot struct

+ MsgPerMin float64

+ Processing time.Duration

type Client

+ WatchBehaviorFingerprint func(ctx context.Context, sub Subscription, cfg BehaviorFingerprintConfig) (*BehaviorFingerprint, error)

type DurableConsumerConfig

+ Metadata map[string]string

+ type MsgRangeOpt func(*msgRangeConfig)

+ func WithMaxMessages(n int) MsgRangeOpt

type Replay

+ FindFirstSeqAtOrAfter func(ctx context.Context, stream string, t time.Time) (uint64, error)

+ FindLastSeqAtOrBefore func(ctx context.Context, stream string, t time.Time) (uint64, error)

+ GetMsgRange func(ctx context.Context, stream string, startSeq, endSeq uint64, ...) ([]*StoredMessage, bool, error)

+ GetMsgRangeByTime func(ctx context.Context, stream string, start, end time.Time, opts ...MsgRangeOpt) ([]*StoredMessage, bool, error)

type ReplayConfig

+ Limit int

+ UntilSeq uint64

+ UntilTime *time.Time

+ type ReplayConsumerResult struct

+ Durable string

+ Limit int

+ StartSeq uint64

+ StartTime *time.Time

+ UntilSeq uint64

+ UntilTime *time.Time

type ReplayOpt

+ func Limit(n int) ReplayOpt

+ func OneMessage(seq uint64) ReplayOpt

+ func UntilSeq(seq uint64) ReplayOpt

+ func UntilTime(t time.Time) ReplayOpt

+ type SlowConsumer struct

+ func WatchSlowConsumer(ctx context.Context, sub Subscription, streamLastSeq StreamLastSeqFunc, ...) (*SlowConsumer, error)

+ func (s *SlowConsumer) Events() <-chan SlowConsumerEvent

+ func (s *SlowConsumer) Slow() bool

+ func (s *SlowConsumer) Stop()

+ type SlowConsumerConfig struct

+ AckPendingRatio float64

+ CircuitStop bool

+ LagThreshold uint64

+ OnSlow func(SlowConsumerEvent)

+ PendingThreshold uint64

+ PollInterval time.Duration

+ SustainFor time.Duration

+ type SlowConsumerEvent struct

+ AckPending int

+ Durable string

+ Lag uint64

+ MaxAckPending int

+ Pending uint64

+ Reasons []string

+ Stream string

+ SustainedFor time.Duration

+ type StoredMessage struct

+ Data []byte

+ Header map[string][]string

+ MessageType MessageType

+ Sequence uint64

+ Subject string

+ Time time.Time

+ type StreamLastSeqFunc func(ctx context.Context, stream string) (uint64, error)

v0.2.1

Jul 25, 2026

Changes in this version

type DurableConsumerConfig

+ HasAckPolicy bool

+ HasDeliverPolicy bool

+ HasReplayPolicy bool

v0.2.0

Jul 25, 2026

Changes in this version

+ const AckAll

+ const AckExplicit

+ const AckNone

+ const ContentTypeJSON

+ const ContentTypeMsgPack

+ const ContentTypeProto

+ const DeliverAll

+ const DeliverByStartSequence

+ const DeliverByStartTime

+ const DeliverLast

+ const DeliverLastPerSubject

+ const DeliverNew

+ const DiscardNew

+ const DiscardOld

+ const FileStorage

+ const HeaderAutopsyError

+ const HeaderAutopsyHash

+ const HeaderAutopsyStack

+ const HeaderContentType

+ const HeaderDLQConsumer

+ const HeaderDLQNumDelivered

+ const HeaderDLQOriginalSubject

+ const HeaderDLQReason

+ const HeaderDLQSequence

+ const HeaderDLQStream

+ const HeaderMsgID

+ const HeaderTraceID

+ const InterestPolicy

+ const LimitsPolicy

+ const MemoryStorage

+ const ReplayInstant

+ const ReplayOriginal

+ const WorkQueuePolicy

+ var ErrAsyncPublishPendingLimit = errors.New("async publish pending limit exceeded")

+ var ErrBackpressureHandled = errors.New("message handled by backpressure policy")

+ var ErrConflictingAuth = errors.New(...)

+ var ErrConsumerRecreateRequired = errors.New(...)

+ var ErrConsumerStall = errors.New("consumer soft-liveness stall: pending rising without process activity")

+ var ErrDLQRouted = dlq.ErrDLQRouted

+ var ErrDrainTimeout = errors.New("connection drain timed out")

+ var ErrEmptyAddressNotAllowed = errors.New("empty address not allowed")

+ var ErrEmptyConfigNotAllowed = errors.New("empty config not allowed")

+ var ErrEmptySubjectNotAllowed = errors.New("empty subject not allowed")

+ var ErrInvalidBucketName = errors.New("invalid kv bucket name")

+ var ErrInvalidDurableName = errors.New("invalid durable name")

+ var ErrInvalidKVKey = errors.New("invalid kv key")

+ var ErrInvalidMessageType = errors.New("invalid message type")

+ var ErrInvalidNKeySeed = errors.New("invalid nkey seed")

+ var ErrInvalidQueueName = errors.New("invalid queue name")

+ var ErrInvalidStreamName = errors.New("invalid stream name")

+ var ErrInvalidSubject = errors.New("invalid subject")

+ var ErrInvalidSubscription = errors.New("invalid subscription")

+ var ErrInvalidTypeAssertion = errors.New("invalid type assertion")

+ var ErrJetStreamV2Required = errors.New("operation requires jetstream v2 API")

+ var ErrNatsConnectionNotEstablished = errors.New("NATS connection is not established")

+ var ErrPoolFull = errors.New("worker pool queue full")

+ var ErrSendToDLQ = dlq.ErrSendToDLQ

+ var ErrSupervisorGiveUp = errors.New("subscription supervisor gave up after max retries")

+ func Decode(data []byte, typ MessageType, dst any) error

+ func DecodeMsg(msg *natspkg.Msg, typ MessageType, dst any) error

+ func DecodeProto(data []byte, msg proto.Message) error

+ func DecodeTyped[T any](msg *natspkg.Msg, typ MessageType) (T, error)

+ func Encode(msg Message) ([]byte, error)

+ func InProgress(msg *natspkg.Msg) error

+ func ListStreams(ctx context.Context, s StreamManager) ([]*natspkg.StreamInfo, error)

+ func NakWithDelay(msg *natspkg.Msg, delay time.Duration) error

+ func RespondBytes(msg *natspkg.Msg, data []byte) error

+ func RespondJSON(msg *natspkg.Msg, v any) error

+ func RespondMsgPack(msg *natspkg.Msg, v any) error

+ func RespondProto(msg *natspkg.Msg, v proto.Message) error

+ func ShardIndex(key string, numShards int) int

+ func ShardSubject(prefix, key string, numShards int, action string) string

+ func StreamNames(ctx context.Context, s StreamManager) ([]string, error)

+ func SupervisePullProcess(ctx context.Context, cfg SupervisorConfig, metrics *clientMetrics, ...) error

+ func TermWithReason(msg *natspkg.Msg, reason string) error

+ func TraceIDFromHeader(h natspkg.Header) string

+ func ValidateBucketName(name string) error

+ func ValidateDurableName(name string) error

+ func ValidateKVKey(key string) error

+ func ValidatePublishSubject(subject string) error

+ func ValidateQueueName(name string) error

+ func ValidateStreamName(name string) error

+ func ValidateSubject(subject string) error

+ func ValidateSubjects(subjects []string) error

+ type AckPolicy = natspkg.AckPolicy

+ type AutopsyConfig = dlq.AutopsyConfig

+ type BackpressureConfig struct

+ MaxAckPending int

+ Mode BackpressureMode

+ PendingMsgBuffer int

+ PendingMsgLimit int

+ QueueDepthSampleInterval time.Duration

+ type BackpressureMode uint8

+ const BackpressureBlock

+ const BackpressureDrop

+ const BackpressureNak

+ const BackpressureTerm

+ type Client interface

+ Connector func() Connector

+ Consumer func() Consumer

+ Consumers func() ConsumerManager

+ KV func() KeyValueManager

+ KVKeys func() KeyValueKeys

+ Monitoring func() Monitoring

+ Objects func() ObjectStoreManager

+ PublishRaw func(ctx context.Context, subject string, data []byte, headers map[string]string) (*PubAck, error)

+ Publisher func() Publisher

+ Replay func() Replay

+ Requester func() Requester

+ Responder func() Responder

+ SetupWorker func(ctx context.Context, setup WorkerSetup, handler MsgHandler) (Subscription, error)

+ Streams func() StreamManager

+ SupervisePullProcess func(ctx context.Context, stream, durable string, handler MsgHandler, ...) error

+ SuperviseQueueSubscribeBound func(ctx context.Context, stream, durable, queue, subject string, ...) (SupervisedSubscription, error)

+ SuperviseSubscribeBound func(ctx context.Context, stream, durable, subject string, handler MsgHandler, ...) (SupervisedSubscription, error)

+ WatchSoftLiveness func(ctx context.Context, sub Subscription, cfg SoftLivenessConfig) (*SoftLiveness, error)

+ WithShadow func(cfg ShadowConfig, primary, shadowHandler MsgHandler) MsgHandler

+ func NewClient(ctx context.Context, cfg *Config) (Client, error)

+ type Config struct

+ Backpressure BackpressureConfig

+ Conn Connection

+ Metrics MetricsConfig

+ PublisherConfig PublisherConfig

+ RequesterConfig RequesterConfig

+ ResponderConfig ResponderConfig

+ RuntimeConsumer RuntimeConsumerConfig

+ Stream StreamConfig

+ func DefaultConfig() Config

+ func DevConfig() Config

+ func ProdFanOutConfig() Config

+ func ProdWorkerConfig() Config

+ func ThroughputConfig() Config

+ type Connection struct

+ Address string

+ AllowMetrics bool

+ AllowReconnect bool

+ ClientName string

+ ConnectTimeout time.Duration

+ CredentialsFile string

+ CustomReconnectDelay func(attempts int) time.Duration

+ DontRandomize bool

+ DrainTimeout time.Duration

+ FlusherTimeout time.Duration

+ HealthCheckInterval time.Duration

+ InitialRetryAttempts int

+ InitialRetryBackoff float64

+ InitialRetryWait time.Duration

+ MaxPingsOut int

+ MaxReconnect int

+ MetricPrefix string

+ OnClosed func(*natspkg.Conn)

+ OnDisconnect func(*natspkg.Conn, error)

+ OnReconnect func(*natspkg.Conn)

+ Password string

+ PingInterval time.Duration

+ ReconnectBufSize int

+ ReconnectJitter time.Duration

+ ReconnectJitterTLS time.Duration

+ ReconnectWait time.Duration

+ RetryOnFailedConnect bool

+ Secret string

+ Seed string

+ TLS ConnectionTLS

+ User string

+ type ConnectionStatus struct

+ Connected bool

+ InLameDuck bool

+ LastDisconnect time.Time

+ LastError error

+ ReconnectCount int64

+ ServerURL string

+ type ConnectionTLS struct

+ CA []byte

+ Cert []byte

+ Config *tls.Config

+ InsecureSkipVerify bool

+ Key []byte

+ ServerName string

+ type Connector interface

+ AccountInfo func(ctx context.Context) (*natspkg.AccountInfo, error)

+ Conn func() *natspkg.Conn

+ ConnectionStatus func() ConnectionStatus

+ HealthCheck func(ctx context.Context) error

+ IsConnected func() bool

+ JetStream func() natspkg.JetStreamContext

+ Shutdown func() error

+ WaitConnected func(ctx context.Context) error

+ type Consumer interface

+ Pull func(stream, durable string) (PullConsumer, error)

+ QueueSubscribe func(ctx context.Context, queue, subject string, handler MsgHandler, ...) (Subscription, error)

+ QueueSubscribeBound func(ctx context.Context, stream, durable, queue, subject string, ...) (Subscription, error)

+ Subscribe func(ctx context.Context, subject string, handler MsgHandler, ...) (Subscription, error)

+ SubscribeBound func(ctx context.Context, stream, durable, subject string, handler MsgHandler) (Subscription, error)

+ type ConsumerConfig = RuntimeConsumerConfig

+ type ConsumerManager interface

+ AddConsumer func(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)

+ ConsumerInfo func(ctx context.Context, stream, durable string) (*natspkg.ConsumerInfo, error)

+ ConsumerNames func(ctx context.Context, stream string) ([]string, error)

+ CreateOrUpdateConsumer func(ctx context.Context, stream string, cfg DurableConsumerConfig) (*natspkg.ConsumerInfo, error)

+ DeleteConsumer func(ctx context.Context, stream, durable string) error

+ ListConsumers func(ctx context.Context, stream string) ([]*natspkg.ConsumerInfo, error)

+ ListConsumersPage func(ctx context.Context, stream string, offset, limit int) ([]*natspkg.ConsumerInfo, int, error)

+ PauseConsumer func(ctx context.Context, stream, durable string, pauseUntil time.Time) error

+ ResumeConsumer func(ctx context.Context, stream, durable string) error

+ UpdateConsumer func(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)

+ type DLQConfig struct

+ Autopsy AutopsyConfig

+ MaxDeliver uint64

+ Publisher Publisher

+ Reason string

+ Recorder *FlightRecorder

+ Subject string

+ type DeliverPolicy = natspkg.DeliverPolicy

+ type DiscardPolicy = natspkg.DiscardPolicy

+ type DurableConsumerConfig struct

+ AckPolicy AckPolicy

+ AckWait time.Duration

+ DeliverPolicy DeliverPolicy

+ Durable string

+ FilterSubject string

+ FilterSubjects []string

+ FlowControl bool

+ Heartbeat time.Duration

+ InactiveThreshold time.Duration

+ MaxAckPending int

+ MaxDeliver int

+ MaxWaiting int

+ MemStorage bool

+ OptStartSeq uint64

+ OptStartTime *time.Time

+ RateLimit uint64

+ ReplayPolicy ReplayPolicy

+ Replicas int

+ type FetchOpt func(*fetchConfig)

+ func WithFetchHeartbeat(d time.Duration) FetchOpt

+ func WithFetchMaxWait(d time.Duration) FetchOpt

+ type FlightRecorder struct

+ func NewFlightRecorder(capacity int) *FlightRecorder

+ func (r *FlightRecorder) AttachSoftLiveness(cfg *SoftLivenessConfig)

+ func (r *FlightRecorder) AttachSupervisor(cfg *SupervisorConfig)

+ func (r *FlightRecorder) Events() <-chan IncidentEvent

+ func (r *FlightRecorder) LogSnapshot(ctx context.Context, msg string)

+ func (r *FlightRecorder) Record(ev IncidentEvent)

+ func (r *FlightRecorder) RecordDLQ(subject, stream, consumer, reason string, seq uint64)

+ func (r *FlightRecorder) RecordDLQAutopsy(subject, stream, consumer, reason, errStr string, seq uint64)

+ func (r *FlightRecorder) RecordReconnect(detail string)

+ func (r *FlightRecorder) Snapshot() []IncidentEvent

+ func (r *FlightRecorder) WriteJSON(w io.Writer) error

+ type IncidentEvent struct

+ Attempt int

+ Consumer string

+ Detail string

+ Err string

+ Kind IncidentKind

+ NumPending uint64

+ Reason string

+ Sequence uint64

+ Stream string

+ Subject string

+ Time time.Time

+ type IncidentKind uint8

+ const IncidentDLQ

+ const IncidentReconnect

+ const IncidentShadow

+ const IncidentStall

+ const IncidentSupervisor

+ type KVBucketStatus struct

+ Bucket string

+ History int64

+ Values uint64

+ type KVEntry struct

+ Bucket string

+ Created time.Time

+ Key string

+ Revision uint64

+ Value []byte

+ type KeyValueConfig struct

+ Bucket string

+ Compression bool

+ Description string

+ History uint8

+ MaxBytes int64

+ Replicas int

+ Storage StorageType

+ TTL time.Duration

+ type KeyValueKeys interface

+ DeleteKey func(ctx context.Context, bucket, key string) error

+ Get func(ctx context.Context, bucket, key string) (*KVEntry, error)

+ History func(ctx context.Context, bucket, key string) ([]KVEntry, error)

+ ListKeys func(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)

+ Put func(ctx context.Context, bucket, key string, value []byte) (*KVEntry, error)

+ type KeyValueManager interface

+ BucketInfo func(ctx context.Context, bucket string) (KVBucketStatus, error)

+ CreateOrUpdate func(ctx context.Context, cfg KeyValueConfig) (natspkg.KeyValue, error)

+ CreateRaw func(ctx context.Context, cfg *natspkg.KeyValueConfig) (KVBucketStatus, error)

+ Delete func(ctx context.Context, bucket string) error

+ ListBuckets func(ctx context.Context) ([]KVBucketStatus, error)

+ Open func(ctx context.Context, bucket string) (natspkg.KeyValue, error)

+ type Message struct

+ Data any

+ Expect *PublishExpectation

+ Header map[string][]string

+ MessageType MessageType

+ func (m Message) WithExpectedLastMsgID(id string) Message

+ func (m Message) WithExpectedLastSeq(seq uint64) Message

+ func (m Message) WithExpectedLastSeqPerSubject(seq uint64) Message

+ func (m Message) WithExpectedStream(stream string) Message

+ func (m Message) WithMsgID(id string) Message

+ type MessageType int

+ const JSON

+ const MessagePack

+ const Proto

+ const Raw

+ func MessageTypeFromHeader(h natspkg.Header) MessageType

+ type MetricsConfig struct

+ AllowMetrics bool

+ AllowTracing bool

+ CollectInterval time.Duration

+ FixedCardinality bool

+ Lite bool

+ Prefix string

+ TrackedConsumers []TrackedConsumer

+ TrackedStreams []string

+ type Monitoring interface

+ Fetch func(ctx context.Context, baseURL, path string) ([]byte, error)

+ type MsgHandler func(context.Context, *natspkg.Msg) error

+ func HandlerTyped[T any](mt MessageType, fn func(ctx context.Context, subject string, payload T) error) MsgHandler

+ func WithDLQ(cfg DLQConfig, handler MsgHandler) MsgHandler

+ func WithShadow(cfg ShadowConfig, primary, shadowHandler MsgHandler) MsgHandler

+ type ObjectBucketStatus struct

+ Bucket string

+ Description string

+ Size uint64

+ type ObjectEntry struct

+ Bucket string

+ Data []byte

+ Modified time.Time

+ Name string

+ Size uint64

+ type ObjectStoreConfig struct

+ Bucket string

+ Description string

+ MaxBytes int64

+ Replicas int

+ Storage StorageType

+ TTL time.Duration

+ type ObjectStoreManager interface

+ BucketInfo func(ctx context.Context, bucket string) (ObjectBucketStatus, error)

+ Create func(ctx context.Context, cfg ObjectStoreConfig) (ObjectBucketStatus, error)

+ CreateRaw func(ctx context.Context, cfg *natspkg.ObjectStoreConfig) (ObjectBucketStatus, error)

+ Delete func(ctx context.Context, bucket string) error

+ DeleteObject func(ctx context.Context, bucket, name string) error

+ Get func(ctx context.Context, bucket, name string) (*ObjectEntry, error)

+ ListBuckets func(ctx context.Context) ([]ObjectBucketStatus, error)

+ ListObjects func(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)

+ Put func(ctx context.Context, bucket, name string, data []byte) (*ObjectEntry, error)

+ type ProcessActivity struct

+ func NewProcessActivity() *ProcessActivity

+ func (a *ProcessActivity) LastSuccess() time.Time

+ func (a *ProcessActivity) Touch()

+ type ProcessOpt func(*processConfig)

+ func WithFetchBatch(batch int) ProcessOpt

+ func WithProcessConcurrency(n int) ProcessOpt

+ func WithProcessHeartbeat(d time.Duration) ProcessOpt

+ func WithProcessMaxWait(d time.Duration) ProcessOpt

+ type PubAck = natspkg.PubAck

+ type PubAckFuture = natspkg.PubAckFuture

+ type PublishExpectation struct

+ LastMsgID string

+ LastSeq *uint64

+ LastSeqPerSubject *uint64

+ Stream string

+ type Publisher interface

+ PublishAsync func(ctx context.Context, subject string, msg Message) (PubAckFuture, error)

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

+ PublishAsyncComplete func(ctx context.Context) error

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

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

+ PublishJSON func(ctx context.Context, subject string, data any) error

+ PublishMessage func(ctx context.Context, subject string, msg Message) error

+ PublishMsgPack func(ctx context.Context, subject string, data any) error

+ PublishProto func(ctx context.Context, subject string, payload proto.Message) error

+ PublishWithMsgID func(ctx context.Context, subject, id string, msg Message) error

+ type PublisherConfig struct

+ AllowMetrics bool

+ AllowTracing bool

+ MaxAsyncPending int

+ MetricPrefix string

+ SkipSubjectValidation bool

+ type PullConsumer interface

+ Fetch func(ctx context.Context, batch int, opts ...FetchOpt) ([]*natspkg.Msg, error)

+ FetchNoWait func(batch int) ([]*natspkg.Msg, error)

+ Process func(ctx context.Context, handler MsgHandler, opts ...ProcessOpt) error

+ type PurgeOpt func(*natspkg.StreamPurgeRequest)

+ func PurgeSubject(subject string) PurgeOpt

+ type Replay interface

+ CreateReplayConsumer func(ctx context.Context, stream, sourceDurable string, opts ...ReplayOpt) (durable string, err error)

+ GetLastMsgForSubject func(ctx context.Context, stream, subject string) (*Message, error)

+ GetMsg func(ctx context.Context, stream string, seq uint64) (*Message, error)

+ GetNextMsgAfter func(ctx context.Context, stream string, seq uint64) (*Message, error)

+ ResetConsumer func(ctx context.Context, stream, durable string, opts ...ReplayOpt) error

+ type ReplayConfig struct

+ DeliverPolicy DeliverPolicy

+ Durable string

+ FilterSubject string

+ FilterSubjects []string

+ OptStartSeq uint64

+ OptStartTime *time.Time

+ ReplayPolicy ReplayPolicy

+ type ReplayOpt func(*ReplayConfig)

+ func FromBeginning() ReplayOpt

+ func FromNew() ReplayOpt

+ func FromSeq(seq uint64) ReplayOpt

+ func FromTime(t time.Time) ReplayOpt

+ func WithDeliverPolicy(policy DeliverPolicy) ReplayOpt

+ func WithFilterSubject(subject string) ReplayOpt

+ func WithFilterSubjects(subjects ...string) ReplayOpt

+ func WithReplayDurable(name string) ReplayOpt

+ func WithReplayPolicy(policy ReplayPolicy) ReplayOpt

+ func WithStartSeq(seq uint64) ReplayOpt

+ func WithStartTime(t time.Time) ReplayOpt

+ type ReplayPolicy = natspkg.ReplayPolicy

+ type Requester interface

+ RequestBytes func(ctx context.Context, subject string, data []byte) (*natspkg.Msg, error)

+ RequestJSON func(ctx context.Context, subject string, req any) (*natspkg.Msg, error)

+ RequestJSONInto func(ctx context.Context, subject string, req, resp any) error

+ RequestMessage func(ctx context.Context, subject string, msg Message) (*natspkg.Msg, error)

+ RequestMsgPack func(ctx context.Context, subject string, req any) (*natspkg.Msg, error)

+ RequestMsgPackInto func(ctx context.Context, subject string, req, resp any) error

+ RequestProto func(ctx context.Context, subject string, req proto.Message) (*natspkg.Msg, error)

+ RequestProtoInto func(ctx context.Context, subject string, req, resp proto.Message) error

+ type RequesterConfig struct

+ AllowMetrics bool

+ AllowTracing bool

+ MetricPrefix string

+ SkipSubjectValidation bool

+ Timeout time.Duration

+ type Responder interface

+ QueueSubscribe func(ctx context.Context, queue, subject string, handler MsgHandler) (Subscription, error)

+ Subscribe func(ctx context.Context, subject string, handler MsgHandler) (Subscription, error)

+ type ResponderConfig struct

+ AllowMetrics bool

+ AllowTracing bool

+ MetricPrefix string

+ type RetentionPolicy = natspkg.RetentionPolicy

+ type RuntimeConsumerConfig struct

+ AckWait time.Duration

+ AllowMetrics bool

+ AllowTracing bool

+ FlowControl bool

+ IdleHeartbeat time.Duration

+ MetricPrefix string

+ PendingMsgBuffer int

+ PendingMsgLimit int

+ WorkerBufferSize int

+ WorkerPoolEnabled bool

+ WorkerPoolSize int

+ type ShadowConfig struct

+ Compare func(primaryErr, shadowErr error) bool

+ Recorder *FlightRecorder

+ SampleRate float64

+ type SoftLiveness struct

+ func WatchSoftLiveness(ctx context.Context, sub Subscription, activity *ProcessActivity, ...) (*SoftLiveness, error)

+ func (s *SoftLiveness) Activity() *ProcessActivity

+ func (s *SoftLiveness) Events() <-chan SoftLivenessEvent

+ func (s *SoftLiveness) Stalled() bool

+ func (s *SoftLiveness) Stop()

+ type SoftLivenessConfig struct

+ CircuitStop bool

+ OnStall func(SoftLivenessEvent)

+ PollInterval time.Duration

+ RisingWindows int

+ StallAfter time.Duration

+ type SoftLivenessEvent struct

+ Err error

+ NumPending uint64

+ StalledFor time.Duration

+ type StorageType = natspkg.StorageType

+ type StreamConfig struct

+ Description string

+ Discard DiscardPolicy

+ DuplicateWindow time.Duration

+ MaxAge time.Duration

+ MaxBytes int64

+ MaxConsumers int

+ MaxMsgSize int32

+ MaxMsgs int64

+ Mirror *StreamSource

+ Name string

+ NoAck bool

+ Replicas int

+ Retention RetentionPolicy

+ Sources []*StreamSource

+ Storage StorageType

+ Subjects []string

+ type StreamManager interface

+ AddStream func(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)

+ CreateOrUpdateStream func(ctx context.Context, cfg StreamConfig) (*natspkg.StreamInfo, error)

+ DeleteStream func(ctx context.Context, name string) error

+ GetLastMsg func(ctx context.Context, stream, subject string) (*natspkg.RawStreamMsg, error)

+ GetMsg func(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)

+ GetNextMsgAfter func(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)

+ ListStreamsPage func(ctx context.Context, offset, limit int) ([]*natspkg.StreamInfo, int, error)

+ PurgeStream func(ctx context.Context, name string, opts ...PurgeOpt) error

+ StreamInfo func(ctx context.Context, name string) (*natspkg.StreamInfo, error)

+ UpdateStream func(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)

+ type StreamSource = natspkg.StreamSource

+ type SubscribeFn func(ctx context.Context) (Subscription, error)

+ type Subscription interface

+ ConsumerInfo func() (*natspkg.ConsumerInfo, error)

+ Drain func() error

+ IsValid func() bool

+ SetPendingLimits func(msgLimit, bytesLimit int) error

+ Subject func() string

+ Type func() natspkg.SubscriptionType

+ Unsubscribe func() error

+ type SupervisedSubscription interface

+ Events func() <-chan SupervisorEvent

+ Stop func() error

+ func Supervise(ctx context.Context, cfg SupervisorConfig, metrics *clientMetrics, ...) (SupervisedSubscription, error)

+ type SupervisorConfig struct

+ CheckInterval time.Duration

+ HealthyBackoffInterval time.Duration

+ HealthyChecksBeforeBackoff int

+ InitialBackoff time.Duration

+ MaxBackoff time.Duration

+ MaxRetries int

+ OnEvent func(SupervisorEvent)

+ type SupervisorEvent struct

+ Attempt int

+ Err error

+ Kind SupervisorEventKind

+ type SupervisorEventKind uint8

+ const SupervisorGiveUp

+ const SupervisorInvalid

+ const SupervisorResubscribed

+ type TrackedConsumer struct

+ Durable string

+ Stream string

+ type WorkerSetup struct

+ Consumer DurableConsumerConfig

+ Queue string

+ Stream StreamConfig

+ Subject string

v0.1.1

Jul 24, 2026

v0.1.0

Jul 24, 2026

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL