Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

consumer

package

Go to main page

Versions in this module

v1

v1.7.6

Mar 28, 2026

v1.7.5

Feb 10, 2026

v1.7.4

Feb 9, 2026

v1.7.3

Feb 2, 2026

v1.7.2

Feb 1, 2026

Changes in this version

+ func GetPartitionFromEnv() (int, error)

+ func ParseStatefulSetOrdinal(hostname string) (int, error)

v1.7.1

Feb 1, 2026

Changes in this version

type Broadcast

+ func (b *Broadcast) Stop(ctx context.Context) error

type Dynamic

+ func (d *Dynamic) Stop(ctx context.Context) error

v1.7.0

Feb 1, 2026

Changes in this version

+ const DefaultWIPMinInterval

+ func WithIteratorEscalation(window time.Duration, threshold int) interface

+ func WithIteratorFactory(...) interface

+ type Broadcast struct

+ func NewBroadcast(js jetstream.JetStream, streamName, consumerPrefix, filterSubject string, ...) (*Broadcast, error)

+ func (b *Broadcast) Close(ctx context.Context) error

+ func (b *Broadcast) Start(ctx context.Context) error

+ func (b *Broadcast) UpdateWorkerConsumer(ctx context.Context, workerID string, partitions []types.Partition) error

+ type BroadcastConfig struct

+ ConsumerPrefix string

+ FilterSubject string

+ InstanceID string

+ IteratorEscalationThreshold int

+ IteratorEscalationWindow time.Duration

+ IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

+ Retry RetryConfig

+ StreamName string

+ func (c *BroadcastConfig) SetDefaults() error

+ func (c *BroadcastConfig) Validate() error

+ type BroadcastOption interface

+ func WithInstanceID(id string) BroadcastOption

+ type CommonConfig struct

+ AckPolicy jetstream.AckPolicy

+ AckWait time.Duration

+ BatchSize int

+ FetchTimeout time.Duration

+ InactiveThreshold time.Duration

+ Logger types.Logger

+ ManualAck bool

+ MaxAckPending int

+ MaxDeliver int

+ MaxWaiting int

+ Metrics types.MetricsCollector

+ func (c *CommonConfig) SetDefaults() error

+ func (c *CommonConfig) Validate() error

+ type Dynamic struct

+ func NewDynamic(js jetstream.JetStream, streamName, consumerPrefix, subjectTemplate string, ...) (*Dynamic, error)

+ func (d *Dynamic) Close(ctx context.Context) error

+ func (d *Dynamic) SetResolverMetrics(m subscription.ResolverMetrics)

+ func (d *Dynamic) Update(ctx context.Context, workerID string, partitions []types.Partition) error

+ func (d *Dynamic) UpdateWorkerConsumer(ctx context.Context, workerID string, partitions []types.Partition) error

+ type DynamicConfig struct

+ AllowWorkerIDChange bool

+ ConsumerPrefix string

+ DrainOnRemove bool

+ DrainOnRemoveTimeout time.Duration

+ IteratorEscalationThreshold int

+ IteratorEscalationWindow time.Duration

+ IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

+ MaxConcurrentSubjects int

+ PartitionRefreshMinInterval time.Duration

+ ProcessingGate *subscription.ProcessingGateConfig

+ PullGatingEnabled bool

+ Resolver subscription.ResolverConfig

+ Retry RetryConfig

+ StreamName string

+ SubjectTemplate string

+ func (c *DynamicConfig) SetDefaults() error

+ func (c *DynamicConfig) Validate() error

+ type DynamicOption interface

+ func WithAllowWorkerIDChange(enabled bool) DynamicOption

+ func WithDrainOnRemove(enabled bool, timeout time.Duration) DynamicOption

+ func WithMaxConcurrentSubjects(n int) DynamicOption

+ func WithPartitionRefreshMinInterval(d time.Duration) DynamicOption

+ func WithProcessingGate(cfg *subscription.ProcessingGateConfig) DynamicOption

+ func WithPullGating(enabled bool) DynamicOption

+ func WithResolver(cfg subscription.ResolverConfig) DynamicOption

+ type MessageHandler interface

+ Handle func(ctx context.Context, msg jetstream.Msg) error

+ func NewWIPHandler(handler MessageHandler, cfg WIPConfig) MessageHandler

+ type MessageHandlerFunc func(ctx context.Context, msg jetstream.Msg) error

+ func (f MessageHandlerFunc) Handle(ctx context.Context, msg jetstream.Msg) error

+ type Option interface

+ func WithAckPolicy(p jetstream.AckPolicy) Option

+ func WithAckWait(d time.Duration) Option

+ func WithBatchSize(n int) Option

+ func WithFetchTimeout(d time.Duration) Option

+ func WithInactiveThreshold(d time.Duration) Option

+ func WithLogger(l types.Logger) Option

+ func WithManualAck(enabled bool) Option

+ func WithMaxAckPending(n int) Option

+ func WithMaxDeliver(n int) Option

+ func WithMaxWaiting(n int) Option

+ func WithMetrics(m types.MetricsCollector) Option

+ func WithRetry(cfg RetryConfig) Option

+ type Queue struct

+ func NewQueue(js jetstream.JetStream, streamName, consumerName, filterSubject string, ...) (*Queue, error)

+ func (q *Queue) Start(ctx context.Context) error

+ func (q *Queue) Stop(ctx context.Context) error

+ type QueueConfig struct

+ ConsumerName string

+ FilterSubject string

+ IteratorEscalationThreshold int

+ IteratorEscalationWindow time.Duration

+ IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)

+ Retry RetryConfig

+ StreamName string

+ func DefaultQueueConfig() QueueConfig

+ func (c *QueueConfig) SetDefaults() error

+ func (c *QueueConfig) Validate() error

+ type QueueOption interface

+ type RetryConfig struct

+ Backoff time.Duration

+ Base time.Duration

+ Max time.Duration

+ Multiplier float64

+ Seed int64

+ type Static struct

+ func NewStatic(js jetstream.JetStream, streamName, consumerName, subjectPattern string, ...) (*Static, error)

+ func (s *Static) Partition() int

+ func (s *Static) Start(ctx context.Context) error

+ func (s *Static) Stop(ctx context.Context) error

+ func (s *Static) Subject() string

+ type StaticConfig struct

+ ConsumerName string

+ HashSeed uint64

+ NumPartitions int

+ Partition int

+ StreamName string

+ SubjectPattern string

+ func (c *StaticConfig) SetDefaults() error

+ func (c *StaticConfig) Validate() error

+ type StaticOption interface

+ func WithDispatchByKey() StaticOption

+ func WithHashSeed(seed uint64) StaticOption

+ func WithKeyChannelBuffer(size int) StaticOption

+ func WithKeyExtractor(fn func(msg jetstream.Msg) string) StaticOption

+ func WithKeyIdleTimeout(d time.Duration) StaticOption

+ type WIPConfig struct

+ Interval time.Duration

+ Logger types.Logger

+ MinInterval time.Duration

+ type WIPHandler struct

+ func (w *WIPHandler) Handle(ctx context.Context, msg jetstream.Msg) error

Other modules containing this package

github.com/arloliu/parti/v2