nats

package module
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Jul 25, 2026 License: Apache-2.0 Imports: 35 Imported by: 0

README

nats

JetStream client for Go. Telemetry is optional via tel.

Module: github.com/gopherust-io/nats

OpenSSF Scorecard

Mental model: the app connects and binds. Ops owns streams (CLI/platform). If no stream captures the subject, the message is not retained—that is topology, not a client bug.

go get github.com/gopherust-io/nats@latest
make nats-up   # local broker

# production topology — not NewClient
nats stream add ORDERS --subjects 'orders.>' --storage file --retention workqueue

Labs may call Streams() / Consumers() / SetupWorker explicitly. See API reference. cfg.Stream on NewClient does not create anything.

Worker sketch

package main

import (
	"context"
	"log"

	libnats "github.com/gopherust-io/nats"
	"github.com/gopherust-io/tel"
	natspkg "github.com/nats-io/nats.go"
)

func main() {
	ctx := context.Background()

	telem := tel.NewWithConfig(tel.DefaultDebugConfig()) // prod: DefaultConfig + collector
	_ = telem.Start(ctx)
	defer telem.Shutdown(ctx)
	ctx = tel.WrapContext(ctx, telem)

	cfg := libnats.DefaultConfig()
	cfg.Conn.Address = "nats://127.0.0.1:4222"

	client, err := libnats.NewClient(ctx, &cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Connector().Shutdown()

	// Stream must already exist (CLI above). Then:
	if err := client.Publisher().PublishJSON(ctx, "orders.created", map[string]any{"id": "42"}); err != nil {
		log.Fatal(err)
	}

	handler := func(ctx context.Context, msg *natspkg.Msg) error {
		// nil → Ack; non-nil → Nak / retry
		return nil
	}

	_, err = client.Consumer().QueueSubscribeBound(
		ctx,
		"ORDERS",           // stream
		"orders-processor", // durable
		"orders-workers",   // queue
		"orders.>",         // filter
		handler,
	)
	if err != nil {
		log.Fatal(err)
	}

	select {}
}

Publish variants: PublishBytes, PublishWithMsgID (requires stream DuplicateWindow + stable IDs). Pick one codec per stream and stay consistent.

Pull

Create the pull durable first (CLI or Consumers().CreateOrUpdateConsumer), then:

pull, err := client.Consumer().Pull("ORDERS", "orders-puller")
if err != nil {
	return err
}
return pull.Process(ctx, handler)

Request / Reply (core NATS)

JetStream is not involved. Requester calls Request*; Responder uses core subscribe (no auto Ack/Nak). Handler must Respond*. Codecs match Publisher: Bytes, JSON, MessagePack, Proto.

_, err = client.Responder().Subscribe(ctx, "echo", func(_ context.Context, msg *natspkg.Msg) error {
	return libnats.RespondJSON(msg, map[string]any{"ok": true})
})

var resp map[string]any
err = client.Requester().RequestJSONInto(ctx, "echo", map[string]any{"ping": 1}, &resp)

Also: RequestBytes / RequestMsgPackInto / RequestProtoInto, and RespondBytes / RespondMsgPack / RespondProto. Queue groups: Responder().QueueSubscribe. Default request timeout is 2s when ctx has no deadline (RequesterConfig.Timeout).

Do not request on subjects captured by a JetStream stream—the server PubAck is delivered to the reply inbox. Auth that blocks _INBOX.> also breaks requests — see devops.

Presets

Preset When
DefaultConfig() Starting point; resilient reconnect
DevConfig() Local; quieter reconnect, metrics off
ProdWorkerConfig() Job queue; worker pool + Nak backpressure
ProdFanOutConfig() Event bus; block backpressure
ThroughputConfig() Max throughput; minimal observability

Presets never invent stream topology.

Extensions

Need Package / API
Consume dedup nats/idempotency
Dead letter nats/dlq
Shadow / canary nats/shadow
Sharding nats/shard
Worker pool nats/workerpool
Replay client.Replay()
Supervised sub SuperviseQueueSubscribeBound, SupervisePullProcess

Footguns

Symptom Cause Fix
Publish “works”, nothing stored Subject not captured by any stream Fix stream Subjects / CLI
Wrong or missing deliveries Stream / durable / filter mismatch Use *SubscribeBound
Poison keeps disappearing Handler returns nil after failure Return error to Nak
Stream missing after deploy Relied on cfg.Stream in NewClient CLI or Streams().CreateOrUpdateStream
Request gets {"stream":...} JSON Subject captured by a JetStream stream Use core-only subjects for R/R
Consumer update rejected Deliver policy change on existing durable Recreate durable; see consumer docs

Development

make nats-up
make demo-nats    # ROLE=all|publisher|worker|puller
make test

Docs

Guide When
Getting started Clone → Docker → demos
JetStream guide Publish / retain / consume
Recipes Production configs
Push vs pull Delivery model
Performance Codecs, AttrCache, throughput
Local Docker Single / cluster / supercluster
API reference Binding, middleware, presets

Contributing · Changelog

License

Apache License 2.0 — see LICENSE.

Documentation

Index

Examples

Constants

View Source 
const (
	HeaderDLQOriginalSubject = dlq.HeaderOriginalSubject
	HeaderDLQStream          = dlq.HeaderStream
	HeaderDLQSequence        = dlq.HeaderSequence
	HeaderDLQNumDelivered    = dlq.HeaderNumDelivered
	HeaderDLQReason          = dlq.HeaderReason
	HeaderDLQConsumer        = dlq.HeaderConsumer

	HeaderAutopsyError = dlq.HeaderAutopsyError
	HeaderAutopsyHash  = dlq.HeaderAutopsyHash
	HeaderAutopsyStack = dlq.HeaderAutopsyStack
)

DLQ header keys (re-exported from nats/dlq).

View Source 
const (
	HeaderContentType  = "Nats-Content-Type"
	HeaderMsgID        = "Nats-Msg-Id"
	HeaderTraceID      = "Trace-Id"
	ContentTypeJSON    = "json"
	ContentTypeProto   = "protobuf"
	ContentTypeMsgPack = "msgpack"
)
View Source 
const (
	DeliverAll             = natspkg.DeliverAllPolicy
	DeliverLast            = natspkg.DeliverLastPolicy
	DeliverNew             = natspkg.DeliverNewPolicy
	DeliverByStartSequence = natspkg.DeliverByStartSequencePolicy
	DeliverByStartTime     = natspkg.DeliverByStartTimePolicy
	DeliverLastPerSubject  = natspkg.DeliverLastPerSubjectPolicy

	ReplayInstant  = natspkg.ReplayInstantPolicy
	ReplayOriginal = natspkg.ReplayOriginalPolicy

	AckExplicit = natspkg.AckExplicitPolicy
	AckNone     = natspkg.AckNonePolicy
	AckAll      = natspkg.AckAllPolicy

	LimitsPolicy    = natspkg.LimitsPolicy
	InterestPolicy  = natspkg.InterestPolicy
	WorkQueuePolicy = natspkg.WorkQueuePolicy

	FileStorage   = natspkg.FileStorage
	MemoryStorage = natspkg.MemoryStorage

	DiscardOld = natspkg.DiscardOld
	DiscardNew = natspkg.DiscardNew
)

Variables

View Source 
var (
	ErrEmptySubjectNotAllowed       = errors.New("empty subject not allowed")
	ErrEmptyConfigNotAllowed        = errors.New("empty config not allowed")
	ErrEmptyAddressNotAllowed       = errors.New("empty address not allowed")
	ErrInvalidSubscription          = errors.New("invalid subscription")
	ErrNatsConnectionNotEstablished = errors.New("NATS connection is not established")
	ErrInvalidStreamName            = errors.New("invalid stream name")
	ErrInvalidDurableName           = errors.New("invalid durable name")
	ErrInvalidQueueName             = errors.New("invalid queue name")
	ErrInvalidBucketName            = errors.New("invalid kv bucket name")
	ErrInvalidSubject               = errors.New("invalid subject")
	ErrInvalidKVKey                 = errors.New("invalid kv key")
	ErrBackpressureHandled          = errors.New("message handled by backpressure policy")
	ErrDrainTimeout                 = errors.New("connection drain timed out")
	ErrJetStreamV2Required          = errors.New("operation requires jetstream v2 API")
	ErrInvalidNKeySeed              = errors.New("invalid nkey seed")
	ErrConsumerRecreateRequired     = errors.New("consumer recreate required; delete and recreate explicitly to change immutable settings")
	ErrConflictingAuth              = errors.New("conflicting auth: set only one of Seed, User/Password, Secret, or CredentialsFile")
	ErrSupervisorGiveUp             = errors.New("subscription supervisor gave up after max retries")
	ErrConsumerStall                = errors.New("consumer soft-liveness stall: pending rising without process activity")
)
View Source 
var (
	ErrInvalidMessageType       = errors.New("invalid message type")
	ErrInvalidTypeAssertion     = errors.New("invalid type assertion")
	ErrPoolFull                 = errors.New("worker pool queue full")
	ErrAsyncPublishPendingLimit = errors.New("async publish pending limit exceeded")
)
View Source 
var ErrDLQRouted = dlq.ErrDLQRouted

ErrDLQRouted is returned (and treated as success by processMessage) after a message has been published to the DLQ and Term'd.

View Source 
var ErrSendToDLQ = dlq.ErrSendToDLQ

ErrSendToDLQ may be returned from a handler to force DLQ routing + Term.

Functions

func Decode 

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

func DecodeMsg 

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

func DecodeProto 

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

func DecodeTyped 

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

DecodeTyped is a generic helper for typed message handlers.

func Encode 

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

func InProgress 

func InProgress(msg *natspkg.Msg) error

InProgress tells JetStream the handler is still working (extends AckWait).

func ListStreams 

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

ListStreams returns all stream infos (convenience; not on StreamManager interface).

func NakWithDelay 

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

NakWithDelay negatively acknowledges with a server-side redelivery delay.

func RespondBytes 

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

RespondBytes replies with raw bytes.

func RespondJSON 

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

RespondJSON encodes v as JSON and replies (sets content-type header).

func RespondMsgPack 

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

RespondMsgPack encodes v as MessagePack and replies.

func RespondProto 

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

RespondProto encodes v as protobuf and replies.

func ShardIndex 

func ShardIndex(key string, numShards int) int

ShardIndex returns a stable shard index for key in [0, numShards). Prefer importing github.com/gopherust-io/nats/shard directly.

func ShardSubject 

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

ShardSubject builds a subject for keyed sharding. Prefer importing github.com/gopherust-io/nats/shard directly.

func StreamNames 

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

StreamNames returns sorted stream names via a full ListStreamsPage.

func SupervisePullProcess 

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

SupervisePullProcess runs Pull.Process in a loop, restarting after failures with backoff.

func TermWithReason 

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

TermWithReason terminates redelivery and attaches a reason (servers that support it). Empty reason falls back to Term().

func TraceIDFromHeader 

func TraceIDFromHeader(h natspkg.Header) string

TraceIDFromHeader returns the explicit Trace-Id header value, if present.

func ValidateBucketName 

func ValidateBucketName(name string) error

ValidateBucketName checks JetStream KV bucket naming rules (alphanumeric, underscore, hyphen only).

func ValidateDurableName 

func ValidateDurableName(name string) error

ValidateDurableName checks JetStream durable/consumer naming rules (same constraints as stream names).

func ValidateKVKey 

func ValidateKVKey(key string) error

ValidateKVKey checks JetStream KV key naming rules.

func ValidatePublishSubject 

func ValidatePublishSubject(subject string) error

ValidatePublishSubject checks that subject is a valid literal publish subject (no wildcards).

func ValidateQueueName 

func ValidateQueueName(name string) error

ValidateQueueName checks queue group naming rules (same constraints as durables).

func ValidateStreamName 

func ValidateStreamName(name string) error

ValidateStreamName checks JetStream stream naming rules. Names must be non-empty, filesystem-friendly, and must not contain whitespace, '.', '*', '>', '/', or '\'.

func ValidateSubject 

func ValidateSubject(subject string) error

ValidateSubject checks that subject is a valid NATS subject (wildcards allowed).

func ValidateSubjects 

func ValidateSubjects(subjects []string) error

ValidateSubjects validates a list of stream/filter subjects (wildcards allowed).

Types

type AckPolicy 

type AckPolicy = natspkg.AckPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type AutopsyConfig 

type AutopsyConfig = dlq.AutopsyConfig

AutopsyConfig enriches DLQ publishes with forensic headers.

type BackpressureConfig 

type BackpressureConfig struct {
	Mode                     BackpressureMode
	MaxAckPending            int
	PendingMsgLimit          int
	PendingMsgBuffer         int
	QueueDepthSampleInterval time.Duration
}

type BackpressureMode 

type BackpressureMode uint8
const (
	BackpressureBlock BackpressureMode = iota + 1
	BackpressureNak
	BackpressureTerm
	BackpressureDrop
)

type Client 

type Client interface {
	Consumer() Consumer
	Publisher() Publisher
	Requester() Requester
	Responder() Responder
	Connector() Connector
	Streams() StreamManager
	Consumers() ConsumerManager
	KV() KeyValueManager
	KVKeys() KeyValueKeys
	Objects() ObjectStoreManager
	Monitoring() Monitoring
	Replay() Replay
	// PublishRaw publishes bytes with optional headers and returns the JetStream ack.
	PublishRaw(ctx context.Context, subject string, data []byte, headers map[string]string) (*PubAck, error)
	SetupWorker(ctx context.Context, setup WorkerSetup, handler MsgHandler) (Subscription, error)
	SuperviseQueueSubscribeBound(ctx context.Context, stream, durable, queue, subject string, handler MsgHandler, cfg SupervisorConfig) (SupervisedSubscription, error)
	SuperviseSubscribeBound(ctx context.Context, stream, durable, subject string, handler MsgHandler, cfg SupervisorConfig) (SupervisedSubscription, error)
	SupervisePullProcess(ctx context.Context, stream, durable string, handler MsgHandler, cfg SupervisorConfig, opts ...ProcessOpt) error
	WatchSoftLiveness(ctx context.Context, sub Subscription, cfg SoftLivenessConfig) (*SoftLiveness, error)
	// WithShadow wraps handlers and records shadow_* metrics when metrics are enabled.
	WithShadow(cfg ShadowConfig, primary, shadowHandler MsgHandler) MsgHandler
}

func NewClient 

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

type Config 

type Config struct {
	PublisherConfig PublisherConfig
	RequesterConfig RequesterConfig
	ResponderConfig ResponderConfig
	Metrics         MetricsConfig
	RuntimeConsumer RuntimeConsumerConfig
	// Stream is an optional topology template for callers; NewClient does not
	// create or update it. Provision streams via nats CLI / platform ops, or
	// explicitly with Streams().CreateOrUpdateStream / SetupWorker.
	Stream       StreamConfig
	Conn         Connection
	Backpressure BackpressureConfig
}

func DefaultConfig 

func DefaultConfig() Config

func DevConfig 

func DevConfig() Config

DevConfig returns a minimal local-development configuration.

Example

ExampleDevConfig shows a minimal local development NATS configuration. 

cfg := DevConfig()
_ = cfg.Stream.Name

func ProdFanOutConfig 

func ProdFanOutConfig() Config

ProdFanOutConfig returns a production event-bus configuration.

func ProdWorkerConfig 

func ProdWorkerConfig() Config

ProdWorkerConfig returns a production job-queue worker configuration.

func ThroughputConfig 

func ThroughputConfig() Config

ThroughputConfig returns a job-queue preset tuned for max publish/consume throughput. Observability is minimized; prefer Proto/PublishBytes on the application hot path. ReconnectBufSize stays at the resilient default (16 MiB); set Conn.ReconnectBufSize to -1 for fail-fast publishers that must not buffer while disconnected.

type Connection 

type Connection struct {
	// CustomReconnectDelay overrides the built-in capped exponential delay.
	CustomReconnectDelay func(attempts int) time.Duration

	// Optional hooks run after the library's internal connection handlers.
	OnDisconnect func(*natspkg.Conn, error)
	OnReconnect  func(*natspkg.Conn)
	OnClosed     func(*natspkg.Conn)

	Address  string
	User     string
	Password string
	Seed     string
	Secret   string
	// CredentialsFile is a NATS .creds / chained credentials file (operator JWT model).
	// Mutually exclusive with Seed, User/Password, and Secret.
	CredentialsFile string
	MetricPrefix    string

	ClientName string
	// TLS configures secure connections (mTLS / CA verification).
	TLS ConnectionTLS

	ReconnectWait      time.Duration
	MaxReconnect       int
	ConnectTimeout     time.Duration
	DrainTimeout       time.Duration
	PingInterval       time.Duration
	MaxPingsOut        int
	FlusherTimeout     time.Duration
	ReconnectJitter    time.Duration
	ReconnectJitterTLS time.Duration

	// ReconnectBufSize is the outbound buffer while reconnecting.
	// 0 leaves the nats.go library default (8 MiB); -1 disables buffering.
	ReconnectBufSize int

	HealthCheckInterval  time.Duration
	InitialRetryAttempts int
	InitialRetryWait     time.Duration
	InitialRetryBackoff  float64
	RetryOnFailedConnect bool
	AllowReconnect       bool
	AllowMetrics         bool
	// DontRandomize keeps server order when Address is a comma-separated list.
	DontRandomize bool
}

Connection is a configured value type.

goalign:ignore

type ConnectionStatus 

type ConnectionStatus struct {
	LastDisconnect time.Time
	LastError      error
	ServerURL      string
	ReconnectCount int64
	Connected      bool
	InLameDuck     bool
}

ConnectionStatus is a configured value type.

goalign:ignore

type ConnectionTLS 

type ConnectionTLS struct {
	// Config is used as-is when non-nil (takes precedence over PEM fields).
	Config *tls.Config
	// ServerName overrides the hostname used for certificate verification.
	ServerName string
	// CA is a PEM-encoded CA certificate pool.
	CA []byte
	// Cert and Key are a PEM-encoded client certificate pair (mTLS).
	Cert []byte
	Key  []byte
	// InsecureSkipVerify disables server certificate verification (dev only).
	InsecureSkipVerify bool
}

ConnectionTLS holds optional TLS material for NATS connections. Prefer PEM fields for config-file / env wiring; set Config for full control. ConnectionTLS is a configured value type.

goalign:ignore

type Connector 

type Connector interface {
	IsConnected() bool
	Shutdown() error
	HealthCheck(ctx context.Context) error
	ConnectionStatus() ConnectionStatus
	WaitConnected(ctx context.Context) error
	Conn() *natspkg.Conn
	JetStream() natspkg.JetStreamContext
	AccountInfo(ctx context.Context) (*natspkg.AccountInfo, error)
}

type Consumer 

type Consumer interface {
	Subscribe(ctx context.Context, subject string, handler MsgHandler, opts ...natspkg.SubOpt) (Subscription, error)
	QueueSubscribe(ctx context.Context, queue, subject string, handler MsgHandler, opts ...natspkg.SubOpt) (Subscription, error)
	SubscribeBound(ctx context.Context, stream, durable, subject string, handler MsgHandler) (Subscription, error)
	QueueSubscribeBound(ctx context.Context, stream, durable, queue, subject string, handler MsgHandler) (Subscription, error)
	Pull(stream, durable string) (PullConsumer, error)
}

type ConsumerConfig 

type ConsumerConfig = RuntimeConsumerConfig

ConsumerConfig is an alias kept for backward compatibility.

type ConsumerManager 

type ConsumerManager interface {
	CreateOrUpdateConsumer(ctx context.Context, stream string, cfg DurableConsumerConfig) (*natspkg.ConsumerInfo, error)
	AddConsumer(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)
	UpdateConsumer(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)
	DeleteConsumer(ctx context.Context, stream, durable string) error
	ConsumerInfo(ctx context.Context, stream, durable string) (*natspkg.ConsumerInfo, error)
	ConsumerNames(ctx context.Context, stream string) ([]string, error)
	ListConsumers(ctx context.Context, stream string) ([]*natspkg.ConsumerInfo, error)
	ListConsumersPage(ctx context.Context, stream string, offset, limit int) ([]*natspkg.ConsumerInfo, int, error)
	PauseConsumer(ctx context.Context, stream, durable string, pauseUntil time.Time) error
	ResumeConsumer(ctx context.Context, stream, durable string) error
}

type DLQConfig 

type DLQConfig struct {
	Publisher Publisher
	// Recorder receives an IncidentDLQ event after a successful route (optional).
	Recorder *FlightRecorder
	// Subject is the dead-letter publish subject (e.g. "orders.dlq").
	Subject string
	// Reason is stored in HeaderDLQReason when auto-routing on MaxDeliver.
	Reason string
	// Autopsy adds forensic headers (error, payload hash, optional stack).
	Autopsy AutopsyConfig
	// MaxDeliver routes to DLQ when metadata.NumDelivered >= MaxDeliver (0 disables auto route).
	MaxDeliver uint64
}

DLQConfig configures WithDLQ.

type DeliverPolicy 

type DeliverPolicy = natspkg.DeliverPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type DiscardPolicy 

type DiscardPolicy = natspkg.DiscardPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type DurableConsumerConfig 

type DurableConsumerConfig struct {
	OptStartTime      *time.Time
	Durable           string
	FilterSubject     string
	FilterSubjects    []string
	DeliverPolicy     DeliverPolicy
	ReplayPolicy      ReplayPolicy
	AckPolicy         AckPolicy
	MaxDeliver        int
	AckWait           time.Duration
	MaxAckPending     int
	RateLimit         uint64
	Heartbeat         time.Duration
	InactiveThreshold time.Duration
	Replicas          int
	MaxWaiting        int
	OptStartSeq       uint64
	FlowControl       bool
	MemStorage        bool
	// HasAckPolicy marks AckPolicy as intentionally set so AckNone (zero) is preserved.
	HasAckPolicy bool
	// HasDeliverPolicy marks DeliverPolicy as intentionally set (including DeliverAll).
	HasDeliverPolicy bool
	// HasReplayPolicy marks ReplayPolicy as intentionally set (including ReplayInstant).
	HasReplayPolicy bool
}

DurableConsumerConfig is a configured value type.

goalign:ignore

type FetchOpt 

type FetchOpt func(*fetchConfig)

func WithFetchHeartbeat 

func WithFetchHeartbeat(d time.Duration) FetchOpt

WithFetchHeartbeat sets a PullHeartbeat for the fetch request. Must be less than MaxWait; on miss, Fetch returns ErrNoHeartbeat.

func WithFetchMaxWait 

func WithFetchMaxWait(d time.Duration) FetchOpt

type FlightRecorder 

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

FlightRecorder is a ring buffer of incident events for ops storytelling. goalign:ignore

func NewFlightRecorder 

func NewFlightRecorder(capacity int) *FlightRecorder

NewFlightRecorder creates a recorder with the given capacity (default 128).

func (*FlightRecorder) AttachSoftLiveness 

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

AttachSoftLiveness wraps cfg.OnStall to record stall incidents.

func (*FlightRecorder) AttachSupervisor 

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

AttachSupervisor wraps cfg.OnEvent to record supervisor lifecycle events. On SupervisorGiveUp it auto-logs a compact dump.

func (*FlightRecorder) Events 

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

Events is a live stream of newly recorded incidents (buffered; overflows dropped).

func (*FlightRecorder) LogSnapshot 

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

LogSnapshot writes a compact slog dump of the current snapshot.

func (*FlightRecorder) Record 

func (r *FlightRecorder) Record(ev IncidentEvent)

Record appends an incident (drops oldest when full).

func (*FlightRecorder) RecordDLQ 

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

RecordDLQ records a dead-letter route.

func (*FlightRecorder) RecordDLQAutopsy 

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

RecordDLQAutopsy records a dead-letter route with forensic error detail.

func (*FlightRecorder) RecordReconnect 

func (r *FlightRecorder) RecordReconnect(detail string)

RecordReconnect records a reconnect spike for storytelling.

func (*FlightRecorder) Snapshot 

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

Snapshot returns incidents in chronological order (oldest first).

func (*FlightRecorder) WriteJSON 

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

WriteJSON writes a Snapshot as JSON to w.

type IncidentEvent 

type IncidentEvent struct {
	Time       time.Time    `json:"time"`
	Detail     string       `json:"detail,omitempty"`
	Subject    string       `json:"subject,omitempty"`
	Stream     string       `json:"stream,omitempty"`
	Consumer   string       `json:"consumer,omitempty"`
	Reason     string       `json:"reason,omitempty"`
	Err        string       `json:"err,omitempty"`
	Sequence   uint64       `json:"sequence,omitempty"`
	NumPending uint64       `json:"num_pending,omitempty"`
	Attempt    int          `json:"attempt,omitempty"`
	Kind       IncidentKind `json:"kind"`
}

IncidentEvent is one entry in the flight-recorder timeline.

type IncidentKind 

type IncidentKind uint8

IncidentKind classifies flight-recorder events. 

const (
	IncidentSupervisor IncidentKind = iota + 1
	IncidentStall
	IncidentDLQ
	IncidentReconnect
	IncidentShadow
)

type KVBucketStatus 

type KVBucketStatus struct {
	Bucket  string
	Values  uint64
	History int64
}

KVBucketStatus summarizes a Key-Value bucket.

type KVEntry 

type KVEntry struct {
	Created  time.Time
	Bucket   string
	Key      string
	Value    []byte
	Revision uint64
}

KVEntry is a Key-Value entry with payload.

type KeyValueConfig 

type KeyValueConfig struct {
	Bucket      string
	Description string
	TTL         time.Duration
	MaxBytes    int64
	Storage     StorageType
	Replicas    int
	History     uint8 // defaults to 1 when zero
	Compression bool
}

KeyValueConfig configures a JetStream Key-Value bucket. KeyValueConfig is a configured value type.

goalign:ignore

type KeyValueKeys 

type KeyValueKeys interface {
	ListKeys(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)
	Get(ctx context.Context, bucket, key string) (*KVEntry, error)
	Put(ctx context.Context, bucket, key string, value []byte) (*KVEntry, error)
	DeleteKey(ctx context.Context, bucket, key string) error
	History(ctx context.Context, bucket, key string) ([]KVEntry, error)
}

KeyValueKeys provides key-level KV helpers (kept separate to avoid interfacebloat).

type KeyValueManager 

type KeyValueManager interface {
	CreateOrUpdate(ctx context.Context, cfg KeyValueConfig) (natspkg.KeyValue, error)
	CreateRaw(ctx context.Context, cfg *natspkg.KeyValueConfig) (KVBucketStatus, error)
	Open(ctx context.Context, bucket string) (natspkg.KeyValue, error)
	Delete(ctx context.Context, bucket string) error
	ListBuckets(ctx context.Context) ([]KVBucketStatus, error)
	BucketInfo(ctx context.Context, bucket string) (KVBucketStatus, error)
}

KeyValueManager manages JetStream Key-Value buckets.

type Message 

type Message struct {
	Data   any
	Header map[string][]string
	// Expect applies optimistic concurrency PubOpts (ExpectedStream / LastSeq / LastMsgId).
	Expect      *PublishExpectation
	MessageType MessageType
}

func (Message) WithExpectedLastMsgID 

func (m Message) WithExpectedLastMsgID(id string) Message

WithExpectedLastMsgID requires the stream's last message ID to equal id.

func (Message) WithExpectedLastSeq 

func (m Message) WithExpectedLastSeq(seq uint64) Message

WithExpectedLastSeq requires the stream's last sequence to equal seq before accept.

func (Message) WithExpectedLastSeqPerSubject 

func (m Message) WithExpectedLastSeqPerSubject(seq uint64) Message

WithExpectedLastSeqPerSubject requires the subject's last sequence to equal seq.

func (Message) WithExpectedStream 

func (m Message) WithExpectedStream(stream string) Message

WithExpectedStream requires the message to land in the named stream.

func (Message) WithMsgID 

func (m Message) WithMsgID(id string) Message

WithMsgID sets the JetStream deduplication header (requires stream DuplicateWindow).

Example

ExampleMessage_WithMsgID demonstrates publish deduplication header. 

_ = Message{Data: map[string]string{"id": "1"}, MessageType: JSON}.WithMsgID("pay-123")

type MessageType 

type MessageType int
const (
	JSON        MessageType = 1
	Proto       MessageType = 2
	MessagePack MessageType = 3
	// Raw publishes Data as []byte without encoding (used by DLQ passthrough).
	Raw MessageType = 4
)

func MessageTypeFromHeader 

func MessageTypeFromHeader(h natspkg.Header) MessageType

type MetricsConfig 

type MetricsConfig struct {
	Prefix           string
	TrackedStreams   []string
	TrackedConsumers []TrackedConsumer
	CollectInterval  time.Duration
	AllowMetrics     bool
	AllowTracing     bool
	// Lite registers only publish/consume/ack counters (skips JetStream gauges and RTT).
	Lite bool
	// FixedCardinality records hot-path metrics without per-subject attributes.
	FixedCardinality bool
}

MetricsConfig is a configured value type.

goalign:ignore

type Monitoring 

type Monitoring interface {
	// Fetch GETs baseURL+path. baseURL is the per-cluster monitoring root
	// (e.g. http://127.0.0.1:8222); path is typically "/jsz" or "/varz".
	Fetch(ctx context.Context, baseURL, path string) ([]byte, error)
}

Monitoring fetches NATS server monitoring HTTP endpoints (/varz, /jsz, …).

type MsgHandler 

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

func HandlerTyped 

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

HandlerTyped wraps a typed handler with decode logic.

func WithDLQ 

func WithDLQ(cfg DLQConfig, handler MsgHandler) MsgHandler

WithDLQ wraps a handler so poison messages are published to a DLQ subject and Term'd. Prefer importing github.com/gopherust-io/nats/dlq for new code.

func WithShadow 

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

WithShadow runs primary for delivery fate and optionally a shadow handler. Prefer importing github.com/gopherust-io/nats/shadow for new code.

type ObjectBucketStatus 

type ObjectBucketStatus struct {
	Bucket      string
	Description string
	Size        uint64
}

ObjectBucketStatus summarizes an Object Store bucket.

type ObjectEntry 

type ObjectEntry struct {
	Modified time.Time
	Bucket   string
	Name     string
	Data     []byte
	Size     uint64
}

ObjectEntry is an object payload plus metadata.

type ObjectStoreConfig 

type ObjectStoreConfig struct {
	Bucket      string
	Description string
	TTL         time.Duration
	MaxBytes    int64
	Storage     StorageType
	Replicas    int
}

ObjectStoreConfig configures a JetStream Object Store bucket.

type ObjectStoreManager 

type ObjectStoreManager interface {
	ListBuckets(ctx context.Context) ([]ObjectBucketStatus, error)
	Create(ctx context.Context, cfg ObjectStoreConfig) (ObjectBucketStatus, error)
	CreateRaw(ctx context.Context, cfg *natspkg.ObjectStoreConfig) (ObjectBucketStatus, error)
	BucketInfo(ctx context.Context, bucket string) (ObjectBucketStatus, error)
	Delete(ctx context.Context, bucket string) error
	ListObjects(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)
	Get(ctx context.Context, bucket, name string) (*ObjectEntry, error)
	Put(ctx context.Context, bucket, name string, data []byte) (*ObjectEntry, error)
	DeleteObject(ctx context.Context, bucket, name string) error
}

ObjectStoreManager manages JetStream Object Store buckets and objects.

type ProcessActivity 

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

ProcessActivity tracks the last successful message process (Ack or DLQ Term).

func NewProcessActivity 

func NewProcessActivity() *ProcessActivity

NewProcessActivity creates a tracker seeded to now.

func (*ProcessActivity) LastSuccess 

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

LastSuccess returns the time of the last successful process.

func (*ProcessActivity) Touch 

func (a *ProcessActivity) Touch()

Touch records a successful process at the current time.

type ProcessOpt 

type ProcessOpt func(*processConfig)

func WithFetchBatch 

func WithFetchBatch(batch int) ProcessOpt

func WithProcessConcurrency 

func WithProcessConcurrency(n int) ProcessOpt

WithProcessConcurrency sets parallel handler goroutines per fetched batch (default 1).

func WithProcessHeartbeat 

func WithProcessHeartbeat(d time.Duration) ProcessOpt

WithProcessHeartbeat sets PullHeartbeat used by Process fetch loops. When unset, Process uses RuntimeConsumerConfig.IdleHeartbeat if > 0.

func WithProcessMaxWait 

func WithProcessMaxWait(d time.Duration) ProcessOpt

type PubAck 

type PubAck = natspkg.PubAck

PubAck is a JetStream publish acknowledgement.

type PubAckFuture 

type PubAckFuture = natspkg.PubAckFuture

PubAckFuture is a future for an async JetStream publish ack.

type PublishExpectation 

type PublishExpectation struct {
	LastSeq           *uint64
	LastSeqPerSubject *uint64
	Stream            string
	LastMsgID         string
}

PublishExpectation is optimistic concurrency for JetStream publish. Zero LastSeq / LastSeqPerSubject are ignored unless the corresponding Set* helper is used (pointers distinguish "unset" from "expect zero").

type Publisher 

type Publisher interface {
	PublishMessage(ctx context.Context, subject string, msg Message) error
	PublishProto(ctx context.Context, subject string, payload proto.Message) error
	PublishJSON(ctx context.Context, subject string, data any) error
	PublishMsgPack(ctx context.Context, subject string, data any) error
	PublishBytes(ctx context.Context, subject string, data []byte) error
	PublishBytesWithMsgID(ctx context.Context, subject, id string, data []byte) error
	PublishWithMsgID(ctx context.Context, subject, id string, msg Message) error
	PublishAsync(ctx context.Context, subject string, msg Message) (PubAckFuture, error)
	PublishAsyncBytes(ctx context.Context, subject string, data []byte) (PubAckFuture, error)
	PublishAsyncComplete(ctx context.Context) error
}

type PublisherConfig 

type PublisherConfig struct {
	MetricPrefix string
	AllowMetrics bool
	AllowTracing bool
	// SkipSubjectValidation skips per-publish subject validation for trusted static subjects.
	SkipSubjectValidation bool
	// MaxAsyncPending caps in-flight PublishAsync requests (0 = defaultMaxAsyncPending, -1 = unlimited).
	MaxAsyncPending int
}

PublisherConfig is a configured value type.

goalign:ignore

type PullConsumer 

type PullConsumer interface {
	Fetch(ctx context.Context, batch int, opts ...FetchOpt) ([]*natspkg.Msg, error)
	FetchNoWait(batch int) ([]*natspkg.Msg, error)
	Process(ctx context.Context, handler MsgHandler, opts ...ProcessOpt) error
}

type PurgeOpt 

type PurgeOpt func(*natspkg.StreamPurgeRequest)

func PurgeSubject 

func PurgeSubject(subject string) PurgeOpt

type Replay 

type Replay interface {
	// ResetConsumer seeks an existing durable to a new deliver position while
	// preserving ack limits, filters, and other consumer settings when present.
	ResetConsumer(ctx context.Context, stream, durable string, opts ...ReplayOpt) error
	// CreateReplayConsumer creates a side-car durable for backfill. The live
	// sourceDurable is left untouched. When WithReplayDurable is omitted, a
	// unique name is generated from sourceDurable.
	CreateReplayConsumer(ctx context.Context, stream, sourceDurable string, opts ...ReplayOpt) (durable string, err error)
	GetMsg(ctx context.Context, stream string, seq uint64) (*Message, error)
	GetLastMsgForSubject(ctx context.Context, stream, subject string) (*Message, error)
	GetNextMsgAfter(ctx context.Context, stream string, seq uint64) (*Message, error)
}

type ReplayConfig 

type ReplayConfig struct {
	OptStartTime   *time.Time
	Durable        string // target durable for CreateReplayConsumer; ignored by ResetConsumer
	FilterSubject  string
	FilterSubjects []string
	DeliverPolicy  DeliverPolicy
	ReplayPolicy   ReplayPolicy
	OptStartSeq    uint64
	// contains filtered or unexported fields
}

ReplayConfig holds replay seek settings. Prefer ReplayOpt helpers (FromSeq, FromTime, WithReplayPolicy, …) at call sites.

Zero-value JetStream enums (DeliverAll, ReplayInstant, AckNone) are valid; set flags track whether an option explicitly chose a policy or start position.

type ReplayOpt 

type ReplayOpt func(*ReplayConfig)

ReplayOpt configures ResetConsumer / CreateReplayConsumer.

func FromBeginning 

func FromBeginning() ReplayOpt

FromBeginning replays from the first available message (DeliverAll).

func FromNew 

func FromNew() ReplayOpt

FromNew only delivers messages published after the consumer is (re)created.

func FromSeq 

func FromSeq(seq uint64) ReplayOpt

FromSeq seeks by stream sequence (DeliverByStartSequence + OptStartSeq).

func FromTime 

func FromTime(t time.Time) ReplayOpt

FromTime seeks by start timestamp (DeliverByStartTime + OptStartTime).

func WithDeliverPolicy 

func WithDeliverPolicy(policy DeliverPolicy) ReplayOpt

WithDeliverPolicy sets an explicit deliver policy (including DeliverAll).

func WithFilterSubject 

func WithFilterSubject(subject string) ReplayOpt

WithFilterSubject sets a single filter subject for the replay consumer.

func WithFilterSubjects 

func WithFilterSubjects(subjects ...string) ReplayOpt

WithFilterSubjects sets multi-filter subjects for the replay consumer.

func WithReplayDurable 

func WithReplayDurable(name string) ReplayOpt

WithReplayDurable sets the side-car durable name for CreateReplayConsumer. Ignored by ResetConsumer (which uses the durable argument).

func WithReplayPolicy 

func WithReplayPolicy(policy ReplayPolicy) ReplayOpt

WithReplayPolicy sets Instant vs Original timing replay (including ReplayInstant).

func WithStartSeq 

func WithStartSeq(seq uint64) ReplayOpt

WithStartSeq sets OptStartSeq (does not change DeliverPolicy by itself).

func WithStartTime 

func WithStartTime(t time.Time) ReplayOpt

WithStartTime sets OptStartTime (does not change DeliverPolicy by itself).

type ReplayPolicy 

type ReplayPolicy = natspkg.ReplayPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type Requester 

type Requester interface {
	RequestBytes(ctx context.Context, subject string, data []byte) (*natspkg.Msg, error)
	RequestMessage(ctx context.Context, subject string, msg Message) (*natspkg.Msg, error)

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

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

	RequestProto(ctx context.Context, subject string, req proto.Message) (*natspkg.Msg, error)
	RequestProtoInto(ctx context.Context, subject string, req, resp proto.Message) error
}

Requester issues core NATS request/reply calls (not JetStream).

type RequesterConfig 

type RequesterConfig struct {
	MetricPrefix string
	// Timeout is used when ctx has no deadline (0 = defaultRequestTimeout).
	Timeout      time.Duration
	AllowMetrics bool
	AllowTracing bool
	// SkipSubjectValidation skips per-request subject validation for trusted static subjects.
	SkipSubjectValidation bool
}

RequesterConfig configures core NATS request/reply client calls. RequesterConfig is a configured value type.

goalign:ignore

type Responder 

type Responder interface {
	Subscribe(ctx context.Context, subject string, handler MsgHandler) (Subscription, error)
	QueueSubscribe(ctx context.Context, queue, subject string, handler MsgHandler) (Subscription, error)
}

Responder registers core NATS reply handlers (not JetStream; no auto Ack/Nak).

type ResponderConfig 

type ResponderConfig struct {
	MetricPrefix string
	AllowMetrics bool
	AllowTracing bool
}

ResponderConfig configures core NATS reply subscribers. ResponderConfig is a configured value type.

goalign:ignore

type RetentionPolicy 

type RetentionPolicy = natspkg.RetentionPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type RuntimeConsumerConfig 

type RuntimeConsumerConfig struct {
	MetricPrefix     string
	PendingMsgLimit  int
	PendingMsgBuffer int
	WorkerPoolSize   int
	WorkerBufferSize int
	AckWait          time.Duration
	// IdleHeartbeat enables JetStream idle heartbeats on non-queue push
	// subscriptions (and defaults pull Process heartbeats). 0 disables.
	// Queue groups do not support idle heartbeat (nats.go limitation).
	IdleHeartbeat     time.Duration
	WorkerPoolEnabled bool
	// FlowControl enables JetStream flow control on non-queue push when
	// IdleHeartbeat > 0. Pair with IdleHeartbeat (recommended by nats.go).
	FlowControl  bool
	AllowMetrics bool
	AllowTracing bool
}

RuntimeConsumerConfig is a configured value type.

goalign:ignore

type ShadowConfig 

type ShadowConfig struct {
	// Compare returns true when primary and shadow outcomes match.
	// If nil, outcomes match when both succeed or both fail (nil-ness only).
	Compare func(primaryErr, shadowErr error) bool
	// Recorder receives IncidentShadow events on panic / mismatch (optional).
	Recorder *FlightRecorder

	// SampleRate is the fraction of messages sent to the shadow in (0,1].
	// Zero (default) means 0.1 (10% canary). Set explicitly to 1 for always-on dual-run.
	SampleRate float64
	// contains filtered or unexported fields
}

ShadowConfig configures WithShadow.

type SoftLiveness 

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

SoftLiveness watches a subscription for backlog growth without processing activity. goalign:ignore

func WatchSoftLiveness 

func WatchSoftLiveness(
	ctx context.Context,
	sub Subscription,
	activity *ProcessActivity,
	cfg SoftLivenessConfig,
	metrics *clientMetrics,
) (*SoftLiveness, error)

WatchSoftLiveness starts polling sub.ConsumerInfo for stall conditions. Register activity via SoftLiveness.Activity() and consumer.OnProcessSuccess, or call Touch manually.

func (*SoftLiveness) Activity 

func (s *SoftLiveness) Activity() *ProcessActivity

Activity returns the process-activity tracker used by this watcher.

func (*SoftLiveness) Events 

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

Events returns stall notifications (buffered; overflows are dropped).

func (*SoftLiveness) Stalled 

func (s *SoftLiveness) Stalled() bool

Stalled reports whether a stall has been detected (sticky until Stop).

func (*SoftLiveness) Stop 

func (s *SoftLiveness) Stop()

Stop ends the watch loop.

type SoftLivenessConfig 

type SoftLivenessConfig struct {
	OnStall func(SoftLivenessEvent)
	// PollInterval is how often ConsumerInfo is polled (default 2s).
	PollInterval time.Duration
	// StallAfter is the minimum time since last successful process before a stall can fire (default 10s).
	StallAfter time.Duration
	// RisingWindows is consecutive polls with rising NumPending required (default 3).
	RisingWindows int
	// CircuitStop stops the watcher after the first stall (apps may then restart the pod).
	CircuitStop bool
}

SoftLivenessConfig controls WatchSoftLiveness. goalign:ignore

type SoftLivenessEvent 

type SoftLivenessEvent struct {
	Err        error
	NumPending uint64
	StalledFor time.Duration
}

SoftLivenessEvent is emitted when backlog grows without successful processing.

type StorageType 

type StorageType = natspkg.StorageType

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type StreamConfig 

type StreamConfig struct {
	// Mirror configures this stream as a mirror of another stream (geo / DR).
	// Prefer nats CLI or platform ops for complex cross-domain setups; see devops.md.
	Mirror      *StreamSource
	Name        string
	Description string
	Subjects    []string
	// Sources aggregates messages from other streams into this stream.
	Sources         []*StreamSource
	Storage         StorageType
	MaxBytes        int64
	MaxAge          time.Duration
	MaxMsgs         int64
	Discard         DiscardPolicy
	DuplicateWindow time.Duration
	MaxConsumers    int
	Retention       RetentionPolicy
	Replicas        int
	MaxMsgSize      int32
	NoAck           bool
}

StreamConfig is a configured value type.

goalign:ignore

type StreamManager 

type StreamManager interface {
	CreateOrUpdateStream(ctx context.Context, cfg StreamConfig) (*natspkg.StreamInfo, error)
	AddStream(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)
	UpdateStream(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)
	DeleteStream(ctx context.Context, name string) error
	StreamInfo(ctx context.Context, name string) (*natspkg.StreamInfo, error)
	ListStreamsPage(ctx context.Context, offset, limit int) ([]*natspkg.StreamInfo, int, error)
	PurgeStream(ctx context.Context, name string, opts ...PurgeOpt) error
	GetMsg(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)
	GetLastMsg(ctx context.Context, stream, subject string) (*natspkg.RawStreamMsg, error)
	GetNextMsgAfter(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)
}

StreamManager manages JetStream streams (≤10 methods for interfacebloat). StreamNames is available via ListStreamsPage; ListStreams is a convenience wrapper implemented outside the interface.

type StreamSource 

type StreamSource = natspkg.StreamSource

StreamSource is a JetStream stream source or mirror origin.

type SubscribeFn 

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

SubscribeFn creates a push subscription (e.g. QueueSubscribeBound).

type Subscription 

type Subscription interface {
	Unsubscribe() error
	Drain() error
	IsValid() bool
	Subject() string
	SetPendingLimits(msgLimit, bytesLimit int) error
	ConsumerInfo() (*natspkg.ConsumerInfo, error)
	Type() natspkg.SubscriptionType
}

type SupervisedSubscription 

type SupervisedSubscription interface {
	Subscription
	Events() <-chan SupervisorEvent
	Stop() error
}

SupervisedSubscription is a push subscription that auto-resubscribes when invalid.

func Supervise 

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

Supervise runs subscribe and keeps the subscription alive by resubscribing when IsValid() becomes false (e.g. after idle-heartbeat miss / ErrConsumerNotActive).

type SupervisorConfig 

type SupervisorConfig struct {
	OnEvent        func(SupervisorEvent)
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	CheckInterval  time.Duration
	// HealthyBackoffInterval is the maximum poll interval after consecutive healthy checks (0 disables).
	HealthyBackoffInterval time.Duration
	// HealthyChecksBeforeBackoff is the number of healthy checks before increasing the poll interval.
	HealthyChecksBeforeBackoff int
	// MaxRetries is resubscribe attempts after interest is lost.
	// 0 means default (10). -1 means unlimited.
	MaxRetries int
}

type SupervisorEvent 

type SupervisorEvent struct {
	Err     error
	Kind    SupervisorEventKind
	Attempt int
}

SupervisorEvent is emitted when a supervised subscription is healed or abandoned. goalign:ignore

type SupervisorEventKind 

type SupervisorEventKind uint8

SupervisorEventKind classifies supervisor lifecycle events. 

const (
	SupervisorResubscribed SupervisorEventKind = iota + 1
	SupervisorGiveUp
	SupervisorInvalid
)

type TrackedConsumer 

type TrackedConsumer struct {
	Stream  string
	Durable string
}

type WorkerSetup 

type WorkerSetup struct {
	Queue    string
	Subject  string
	Stream   StreamConfig
	Consumer DurableConsumerConfig
}

WorkerSetup configures a one-shot stream + durable + queue subscribe.

Example

ExampleWorkerSetup documents one-shot worker setup fields. 

_ = WorkerSetup{
	Stream:   StreamConfig{Name: "ORDERS", Subjects: []string{"orders.>"}},
	Consumer: DurableConsumerConfig{Durable: "orders-processor", FilterSubject: "orders.>"},
	Queue:    "orders-workers",
	Subject:  "orders.>",
}

Source Files

View all Source files

Directories

Path Synopsis
examples
nats command
internal
bytesconv
Package bytesconv provides zero-allocation string↔[]byte conversions.
 Package bytesconv provides zero-allocation string↔[]byte conversions.
tools
loadtest command
Command loadtest publishes and consumes JetStream messages for CPU/memory profiling.
 Command loadtest publishes and consumes JetStream messages for CPU/memory profiling.

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.