consumer

package
v1.7.6 Latest Latest
Warning

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

 Go to latest Published: Mar 28, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package consumer provides unified JetStream consumer types for different consumption patterns.

This package offers a consistent API for four distinct consumer types, each designed for specific use cases. All consumer types use the unified MessageHandler interface.

Choosing a Consumer Type 

Type         Use Case                                Coordination  Lifecycle
[Queue]      Load-balanced workers                   None          Start → Stop
[Static]     StatefulSet fixed partition             None          Start → Stop
[Broadcast]  Fan-out to all instances                None          Start → Stop
[Dynamic]    Manager-assigned partitions (Parti)     Via Manager   Update → Stop

Lifecycle

Each consumer type follows a consistent lifecycle pattern:

Queue and Static (Start/Stop pattern) 

// Create and configure
c, err := consumer.NewQueue(js, "stream", "consumer", "subject.>", handler)
if err != nil { log.Fatal(err) }

// Always stop on exit to avoid goroutine leaks
defer c.Stop(ctx)

// Begin consuming (starts background goroutine)
if err := c.Start(ctx); err != nil { log.Fatal(err) }

// ... application runs ...

Broadcast (Start/Stop pattern) 

c, err := consumer.NewBroadcast(js, "stream", "prefix", "events.>", handler)
if err != nil { log.Fatal(err) }
defer c.Stop(ctx)

if err := c.Start(ctx); err != nil { log.Fatal(err) }

Dynamic (Update/Stop pattern) 

c, err := consumer.NewDynamic(js, "stream", "prefix", "orders.{{.PartitionID}}", handler)
if err != nil { log.Fatal(err) }
defer c.Stop(ctx)

// Consumption starts when Update is called (typically by Parti Manager)
if err := c.Update(ctx, "worker-0", partitions); err != nil { log.Fatal(err) }

Thread Safety

All consumer types are safe for concurrent use. Lifecycle methods (Start, Stop, Update) are serialized internally to prevent race conditions.

Message Handler

All consumer types use the unified MessageHandler interface: 

handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
    // Process message
    return nil  // auto-ack on success
})

By default, returning nil automatically acknowledges the message, and returning an error triggers a negative acknowledgement (NAK). Set ManualAck=true to control acknowledgement explicitly via msg.Ack(), msg.Nak(), or msg.Term().

Error Handling

Constructor errors are validation failures (nil JetStream, missing required fields). These are not exported sentinel errors—use string matching or error wrapping checks.

Runtime errors from lifecycle methods:

Consumer Types

Queue

Load-balanced shared consumer. Use for classic worker queue patterns where each message goes to exactly one instance. Multiple replicas share a single durable consumer name, achieving queue group semantics. 

c, _ := consumer.NewQueue(js, "jobs", "job-workers", "jobs.>", handler)
defer c.Stop(ctx)
_ = c.Start(ctx)

Static

Fixed partition assignment (e.g., StatefulSet ordinal). Use when each pod processes exactly one predetermined partition. 

c, _ := consumer.NewStatic(js, "events", "processor-0", "events.{{partition}}", 10, 0, handler)
defer c.Stop(ctx)
_ = c.Start(ctx)

Dynamic

Manager-assigned partitions (Parti core). Use when partitions are dynamically distributed by a coordination layer. The consumer manages multiple internal per-partition consumers based on assignments. 

c, _ := consumer.NewDynamic(js, "events", "worker", "orders.{{.PartitionID}}.events", handler)
defer c.Stop(ctx)
_ = c.Update(ctx, workerID, partitions)

Broadcast

Fan-out to all instances. Use when every instance must receive every message (caching, audit logs, notifications). Each instance receives a copy of every message matching the wildcard filter.

IMPORTANT: The stream MUST use LimitsPolicy or InterestPolicy. WorkQueuePolicy is incompatible because it delivers each message to exactly one consumer. 

c, _ := consumer.NewBroadcast(js, "events", "cache-updater", "events.>", handler)
defer c.Stop(ctx)
_ = c.Start(ctx)

Options

All constructors accept functional options for customization: 

c, _ := consumer.NewQueue(js, "stream", "consumer", "subject.>", handler,
    consumer.WithLogger(myLogger),
    consumer.WithAckWait(60*time.Second),
    consumer.WithBatchSize(10),
)

Common options (Option) work with all consumer types. Type-specific options (e.g., QueueOption, StaticOption) are enforced at compile time.

Field Mapping

The consumer package uses unified field names in constructors: 

consumerName/Prefix  ConsumerName    ConsumerPrefix   ConsumerPrefix   ConsumerName
subject pattern      SubjectPattern  SubjectTemplate  FilterSubject    FilterSubject

Helpers

The package provides helper functions to assist with partition determination in Kubernetes environments:

Example (BroadcastConsumer)

This example demonstrates how to create a Broadcast consumer where every instance receives every message (fan-out pattern). 

package main

import (
	"context"
	"fmt"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	var js jetstream.JetStream

	handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		fmt.Printf("Cache invalidation: %s\n", string(msg.Data()))
		return nil
	})

	c, err := consumer.NewBroadcast(
		js,
		"CACHE-EVENTS",  // streamName
		"cache-updater", // consumerPrefix
		"cache.>",       // filterSubject (wildcard for all cache events)
		handler,
		consumer.WithInstanceID("pod-abc123"), // unique per instance
	)
	if err != nil {
		fmt.Printf("Failed to create broadcast consumer: %v\n", err)
		return
	}

	ctx := context.Background()

	if err := c.Start(ctx); err != nil {
		fmt.Printf("Failed to start: %v\n", err)
		return
	}

	// Every instance running this code receives all cache.> messages
	// Cleanup
	_ = c.Stop(ctx)

}

Output:
Failed to create broadcast consumer: JetStream context is required
Example (DynamicConsumer)

This example demonstrates how to create a Dynamic consumer that receives partition assignments from the Parti Manager. 

package main

import (
	"context"
	"fmt"

	"github.com/arloliu/parti/consumer"
	"github.com/arloliu/parti/types"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	var js jetstream.JetStream

	handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		fmt.Printf("Processing order event: %s\n", string(msg.Data()))
		return nil
	})

	c, err := consumer.NewDynamic(
		js,
		"ORDERS",                  // streamName
		"worker",                  // consumerPrefix
		"orders.{{.PartitionID}}", // subjectTemplate (Go template syntax)
		handler,
	)
	if err != nil {
		fmt.Printf("Failed to create dynamic consumer: %v\n", err)
		return
	}

	ctx := context.Background()

	// In a real application, the Parti Manager calls Update automatically.
	// Here we demonstrate manual usage:
	partitions := []types.Partition{
		{Keys: []string{"partition-0"}},
		{Keys: []string{"partition-1"}},
	}

	if err := c.Update(ctx, "worker-0", partitions); err != nil {
		fmt.Printf("Failed to update: %v\n", err)
		return
	}

	// Consumer is now processing messages for partition-0 and partition-1
	// Cleanup
	_ = c.Stop(ctx)

}

Output:
Failed to create dynamic consumer: JetStream context is required
Example (MessageHandler)

This example demonstrates how to use the MessageHandlerFunc adapter to convert a function into a MessageHandler. 

package main

import (
	"context"
	"fmt"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	// Simple handler that processes and auto-acks
	simple := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		data := msg.Data()
		// Process data...
		_ = data
		return nil // nil = auto-ack
	})

	// Handler that explicitly signals failure (triggers NAK)
	withError := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		if err := processMessage(msg); err != nil {
			return err // non-nil = auto-NAK
		}
		return nil
	})

	// Handler for manual ack mode (requires WithManualAck option)
	manual := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		if err := processMessage(msg); err != nil {
			_ = msg.Term() // terminal failure, don't redeliver
			return err
		}
		_ = msg.Ack() // explicit ack
		return nil
	})

	// All handlers implement MessageHandler interface
	fmt.Printf("Created %d handlers\n", 3)

	// Use any handler with a consumer constructor
	_ = simple
	_ = withError
	_ = manual
}

func processMessage(_ jetstream.Msg) error {

	return nil
}

Output:
Created 3 handlers
Example (QueueConsumer)

This example demonstrates how to create a Queue consumer for load-balanced message processing across multiple worker instances.

In production, you would obtain js from nats.Connect + jetstream.New. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	// Placeholder - in production, obtain from nats.Connect + jetstream.New
	var js jetstream.JetStream

	handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		fmt.Printf("Processing job: %s\n", string(msg.Data()))
		return nil // auto-ack on success
	})

	q, err := consumer.NewQueue(
		js,
		"JOBS",          // streamName
		"job-processor", // consumerName (shared across replicas)
		"jobs.>",        // filterSubject
		handler,
		consumer.WithAckWait(30*time.Second),
		consumer.WithBatchSize(10),
	)
	if err != nil {
		fmt.Printf("Failed to create queue: %v\n", err)
		return
	}

	ctx := context.Background()

	if err := q.Start(ctx); err != nil {
		fmt.Printf("Failed to start: %v\n", err)
		return
	}

	// Consumer is now processing messages in the background.
	// In a real application, wait for shutdown signal, then:
	_ = q.Stop(ctx)

}

Output:
Failed to create queue: JetStream context is required
Example (StaticConsumer)

This example demonstrates how to create a Static consumer for StatefulSet deployments where each pod handles a fixed partition. 

package main

import (
	"context"
	"fmt"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	var js jetstream.JetStream

	handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		fmt.Printf("Processing event: %s\n", string(msg.Data()))
		return nil
	})

	// Pod ordinal determines partition (e.g., pod-0 → partition 0)
	podOrdinal := 0
	numPartitions := 10

	c, err := consumer.NewStatic(
		js,
		"EVENTS",               // streamName
		"processor-0",          // consumerName (unique per partition)
		"events.{{partition}}", // subjectPattern
		numPartitions,          // total partitions
		podOrdinal,             // this pod's partition
		handler,
	)
	if err != nil {
		fmt.Printf("Failed to create static consumer: %v\n", err)
		return
	}

	ctx := context.Background()

	if err := c.Start(ctx); err != nil {
		fmt.Printf("Failed to start: %v\n", err)
		return
	}

	// This consumer only receives messages for partition 0
	fmt.Printf("Consuming partition %d, subject: %s\n", c.Partition(), c.Subject())

	// Cleanup
	_ = c.Stop(ctx)

}

Output:
Failed to create static consumer: JetStream context is required
Example (WipHandler)

This example demonstrates how to wrap a message handler with automatic heartbeats for long-running processing using WIPHandler.

WIPHandler periodically calls msg.InProgress() to extend the AckWait deadline, preventing JetStream from redelivering messages while processing is still active. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	// Base handler that performs long-running processing
	slowHandler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		// Simulate long-running work (e.g., ML inference, large file processing)
		// This could take 30+ seconds
		fmt.Println("Starting long-running processing...")

		// Normally this would timeout and redeliver, but WIPHandler
		// sends InProgress() every 10 seconds to keep it alive
		// time.Sleep(45 * time.Second)

		fmt.Println("Processing complete")

		return nil
	})

	// Wrap with WIPHandler for automatic heartbeats
	// Interval should be < AckWait/2 (e.g., AckWait=30s, Interval=10s)
	wrappedHandler := consumer.NewWIPHandler(slowHandler, consumer.WIPConfig{
		Interval: 10 * time.Second,
		// Logger: myLogger, // Optional: log heartbeat errors
	})

	// Use with any consumer type
	var js jetstream.JetStream
	_, err := consumer.NewQueue(
		js,
		"LONG-JOBS",
		"slow-processor",
		"jobs.slow.>",
		wrappedHandler,
		consumer.WithAckWait(30*time.Second), // Heartbeat at 10s < 30s/2
	)
	if err != nil {
		fmt.Printf("Failed to create queue: %v\n", err)
		return
	}

}

Output:
Failed to create queue: JetStream context is required
Example (WipHandlerWithDynamic)

This example demonstrates using WIPHandler with Dynamic consumer for partition-aware long-running processing. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/arloliu/parti/consumer"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	// Handler that processes large datasets per partition
	handler := consumer.MessageHandlerFunc(func(ctx context.Context, msg jetstream.Msg) error {
		fmt.Printf("Processing batch for subject: %s\n", msg.Subject())
		// Long-running batch processing...
		return nil
	})

	// Wrap with heartbeats
	wrapped := consumer.NewWIPHandler(handler, consumer.WIPConfig{
		Interval: 15 * time.Second, // AckWait is 60s, so 15s is safe (< 30s)
	})

	var js jetstream.JetStream
	_, err := consumer.NewDynamic(
		js,
		"BATCH-EVENTS",
		"batch-worker",
		"batch.{{.PartitionID}}.events",
		wrapped,
		consumer.WithAckWait(60*time.Second),
	)
	if err != nil {
		fmt.Printf("Failed to create dynamic consumer: %v\n", err)
		return
	}

}

Output:
Failed to create dynamic consumer: JetStream context is required

Index

Examples

Constants

View Source 
const DefaultWIPMinInterval = 100 * time.Millisecond

DefaultWIPMinInterval is the minimum allowed heartbeat interval. Intervals below this threshold are clamped to prevent excessive overhead.

Variables

This section is empty.

Functions

func GetPartitionFromEnv added in v1.7.2 

func GetPartitionFromEnv() (int, error)

GetPartitionFromEnv reads the partition index from environment. Wraps partition.GetPartitionFromEnv.

func ParseStatefulSetOrdinal added in v1.7.2 

func ParseStatefulSetOrdinal(hostname string) (int, error)

ParseStatefulSetOrdinal extracts the ordinal index from a StatefulSet pod hostname.

Wraps partition.ParseStatefulSetOrdinal.

func WithIteratorFactory 

func WithIteratorFactory(f func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)) interface {
	QueueOption
	BroadcastOption
	DynamicOption
}

WithIteratorFactory sets a custom iterator factory (for testing). Supported by: Queue, Broadcast, Dynamic.

func WithRetry 

func WithRetry(cfg RetryConfig) interface {
	QueueOption
	BroadcastOption
	DynamicOption
}

WithRetry sets the retry backoff configuration.

Supported by: Queue, Broadcast, Dynamic. Static consumers use a fixed internal retry and ignore this option.

Types

type Broadcast 

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

Broadcast is a fan-out consumer where every instance receives every message. Uses a unique durable name per instance.

Lifecycle

Create with NewBroadcast, then call Broadcast.Start to begin consuming. Clean up with Broadcast.Stop: 

consumer, err := consumer.NewBroadcast(js, "stream", "cache-updater", "events.>", handler)
if err != nil { log.Fatal(err) }
defer consumer.Stop(ctx)

if err := consumer.Start(ctx); err != nil { log.Fatal(err) }

Stream Requirement

The stream MUST use LimitsPolicy or InterestPolicy. WorkQueuePolicy is incompatible because it delivers each message to exactly one consumer, defeating the fan-out purpose.

Thread Safety

Broadcast is safe for concurrent use. Broadcast.Start and Broadcast.Stop are serialized internally.

Deprecation Notice

This type wraps subscription.BroadcastConsumer. Future versions may deprecate the subscription package in favor of this unified consumer API.

func NewBroadcast 

func NewBroadcast(
	js jetstream.JetStream,
	streamName, consumerPrefix, filterSubject string,
	handler MessageHandler,
	opts ...BroadcastOption,
) (*Broadcast, error)

NewBroadcast creates a new broadcast fan-out consumer.

func (*Broadcast) Start 

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

Start begins consuming messages.

The consumer creates a durable JetStream consumer with a unique name derived from the InstanceID and starts a pull loop. All messages matching the FilterSubject are delivered to the handler.

Start may only be called once. Calling Start on an already-started consumer is a no-op.

Parameters:

  • ctx: Context for the start operation. Used for JetStream API calls.

Returns:

  • error: Non-nil if JetStream consumer creation fails.

func (*Broadcast) Stop added in v1.7.1 

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

Stop gracefully stops the consumer.

Stop cancels the internal pull loop and waits for pending message processing to complete (up to the context deadline). The underlying JetStream consumer is NOT deleted; it will be garbage-collected by the server after InactiveThreshold.

Stop is idempotent; calling it multiple times is safe.

Parameters:

  • ctx: Context with shutdown deadline. If the deadline expires, Stop returns context.DeadlineExceeded but the consumer will still eventually stop.

Returns:

  • error: Context error if the wait times out; nil otherwise.

func (*Broadcast) UpdateWorkerConsumer 

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

UpdateWorkerConsumer implements the WorkerConsumerUpdater interface.

For Broadcast consumers, this is equivalent to Broadcast.Start. The workerID and partitions arguments are ignored because Broadcast receives all messages matching the filter regardless of partition assignment.

This method exists for compatibility with code that uses WorkerConsumerUpdater interface.

type BroadcastConfig 

type BroadcastConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// InstanceID identifies the unique instance of the application.
	//
	// This ID allows each instance to have its own durable consumer, ensuring
	// that every instance receives a copy of every message (fan-out).
	//
	// Accepted formats:
	//   - "fixed-string": Uses the literal value as the identity
	//   - "env:ENV_NAME": Uses the value of the specified environment variable
	//
	// If empty, the consumer will attempt to derive an identity from
	// HOSTNAME, then POD_NAME, and finally fall back to a generated short ID.
	InstanceID string

	// FilterSubject is the subject filter to consume from.
	// Supports wildcards (e.g., "orders.*", "events.>").
	// Messages matching this filter will be broadcast to all instances.
	FilterSubject string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	//
	// The final durable name is constructed as "<ConsumerPrefix>_broadcast_<InstanceID>".
	// This ensures that each instance gets a unique durable name, achieving fan-out.
	ConsumerPrefix string `validate:"required"`

	// Retry configures the backoff behavior for control-plane operations
	// (e.g., initial connection, creating the consumer).
	Retry RetryConfig

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)
}

BroadcastConfig configures a Broadcast consumer. Uses unified naming; converted to subscription.BroadcastConsumerConfig internally.

func (*BroadcastConfig) SetDefaults 

func (c *BroadcastConfig) SetDefaults() error

SetDefaults applies default values to the configuration.

func (*BroadcastConfig) Validate 

func (c *BroadcastConfig) Validate() error

Validate checks configuration constraints.

type BroadcastOption 

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

BroadcastOption applies only to Broadcast consumers.

This interface allows both universal Options and Broadcast-specific options (like WithInstanceID) to be passed to NewBroadcast.

func WithInstanceID 

func WithInstanceID(id string) BroadcastOption

WithInstanceID sets the instance ID for broadcast consumers. If unset, hostname or env var is used.

type CommonConfig 

type CommonConfig struct {
	// Logger provides structured logging for the consumer.
	// If nil, a no-op logger is used.
	Logger types.Logger

	// Metrics is the metrics collector for consumer operations.
	// If nil, a no-op collector is used.
	Metrics types.MetricsCollector

	// ManualAck disables automatic acknowledgement of messages.
	//
	// When false (default):
	//  - If the handler returns nil, the message is automatically acknowledged.
	//  - If the handler returns an error, the message is negatively acknowledged (Nak).
	//
	// When true:
	//  - The handler MUST explicitly call msg.Ack(), msg.Nak(), or msg.Term().
	//  - Returning an error from the handler is still logged but does not trigger Action.
	ManualAck bool

	// AckWait is the time allowed for processing a message before it is considered lost
	// and re-delivered by the server.
	//
	// This should be longer than the expected maximum processing time of a single message.
	//
	// Default: 30s.
	AckWait time.Duration `default:"30s" validate:"gt=0"`

	// MaxDeliver is the maximum number of times a message will be delivered.
	//
	// If a message fails processing (Nak) or times out (AckWait) this many times,
	// it will be terminated or moved to a Dead Letter Queue (if configured on the stream).
	//
	// Default: -1 (unlimited).
	MaxDeliver int `default:"-1" validate:"gte=-1"`

	// BatchSize is the maximum number of messages to pull from the server in a single request.
	//
	// A higher batch size can improve throughput but typically increases memory usage
	// and potentially latency for individual messages if processing is slow.
	//
	// Default: 1.
	BatchSize int `default:"1" validate:"gt=0"`

	// FetchTimeout is the maximum duration to wait for a batch of messages to arrive
	// when pulling from the server.
	//
	// If no messages are available within this timeout, the pull request expires.
	// The consumer loop manages this automatically.
	//
	// Default: 5s.
	FetchTimeout time.Duration `default:"5s" validate:"gt=0"`

	// MaxWaiting is the maximum number of outstanding pull requests allowed.
	//
	// This controls the pre-fetch buffer. A value of 2 with BatchSize of 1 means
	// there can be 2 messages buffered locally (1 being processed, 1 ready).
	//
	// Default: 2.
	MaxWaiting int `default:"2" validate:"gt=0"`

	// MaxAckPending limits the number of messages that can be in-flight (unacknowledged)
	// at any given time.
	//
	// If the limit is reached, the server will pause delivery until some messages are acknowledged.
	// If zero, the server's consumer default is used.
	MaxAckPending int `validate:"gte=0"`

	// InactiveThreshold is the duration after which an idle consumer (with no active subscriptions)
	// will be automatically deleted by the server.
	//
	// For durable consumers, this should be set high enough to survive application restarts.
	//
	// Default: 24h.
	InactiveThreshold time.Duration `default:"24h" validate:"gt=0"`

	// AckPolicy controls the JetStream acknowledgement policy.
	//
	// Typically set to AckExplicitPolicy for reliable processing.
	// Defaults to AckExplicitPolicy if usually not set manually.
	AckPolicy jetstream.AckPolicy
}

CommonConfig contains configuration fields shared by all consumer types. Embedding this in each consumer's config ensures consistent naming and defaults.

These settings control the low-level behavior of the JetStream consumer, including acknowledgement policies, batching, timeouts, and redelivery.

func (*CommonConfig) SetDefaults 

func (c *CommonConfig) SetDefaults() error

SetDefaults applies default values to the configuration.

func (*CommonConfig) Validate 

func (c *CommonConfig) Validate() error

Validate checks configuration constraints.

type Dynamic 

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

Dynamic is a partition-aware consumer that receives assignments from a Parti Manager. It manages multiple internal consumers based on assigned partitions.

Lifecycle

Create with NewDynamic, then call Dynamic.Update to start consuming assigned partitions. Clean up with Dynamic.Stop: 

consumer, err := consumer.NewDynamic(js, "stream", "worker", "orders.{{.PartitionID}}", handler)
if err != nil { log.Fatal(err) }
defer consumer.Stop(ctx)

// Start consuming partitions (typically called by Parti Manager)
if err := consumer.Update(ctx, "worker-0", partitions); err != nil { log.Fatal(err) }

Unlike Static and Queue, Dynamic does NOT have a Start method. Consumption begins when Dynamic.Update is called with a non-empty partition list.

Thread Safety

Dynamic is safe for concurrent use. Dynamic.Update calls are serialized internally to prevent race conditions during assignment changes.

Deprecation Notice

This type wraps subscription.WorkerConsumer. Future versions may deprecate the subscription package in favor of this unified consumer API.

func NewDynamic 

func NewDynamic(
	js jetstream.JetStream,
	streamName, consumerPrefix, subjectTemplate string,
	handler MessageHandler,
	opts ...DynamicOption,
) (*Dynamic, error)

NewDynamic creates a new dynamic partition consumer.

func (*Dynamic) SetResolverMetrics 

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

SetResolverMetrics sets the metrics collector for the ownership resolver.

This is an advanced method for observability integration. Most users do not need to call this directly.

func (*Dynamic) Stop added in v1.7.1 

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

Close stops all partition consumers.

Stop gracefully stops all partition consumers.

Stop cancels all internal pull loops and waits for pending message processing to complete (up to the context deadline). The underlying JetStream consumers are NOT deleted; they will be garbage-collected by the server after InactiveThreshold.

If DrainOnRemove is enabled, Stop will first drain pending messages (up to DrainOnRemoveTimeout) before stopping.

Stop is idempotent; calling it multiple times is safe.

Parameters:

  • ctx: Context with shutdown deadline. If the deadline expires, Stop returns context.DeadlineExceeded but consumers will still eventually stop.

Returns:

  • error: Context error if the wait times out; nil otherwise.

func (*Dynamic) Update 

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

Update applies a new partition assignment set.

This method creates or binds durable consumers for newly assigned partitions and stops consumers for removed partitions. The underlying JetStream consumers are NOT deleted on removal; they will be garbage-collected by the server after InactiveThreshold.

Update is typically called by the Parti Manager when assignments change. On the first call, this starts consuming the assigned partitions.

Parameters:

  • ctx: Context for the update operation. Used for JetStream API calls.
  • workerID: The stable worker ID for this instance (e.g., "worker-0").
  • partitions: The new set of partitions to consume. Empty list stops all.

Returns:

Errors:

func (*Dynamic) UpdateWorkerConsumer deprecated 

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

UpdateWorkerConsumer is an alias for Dynamic.Update that implements the WorkerConsumerUpdater interface used by the Parti Manager.

Deprecated: Use Dynamic.Update for new code. This method exists for backward compatibility with code that expects the WorkerConsumerUpdater interface.

type DynamicConfig 

type DynamicConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// SubjectTemplate is a text/template for building subjects from partitions.
	//
	// It relies on the standard Go text/template package.
	// The template context provides a {{.PartitionID}} variable.
	//
	// Example: "orders.{{.PartitionID}}.events"
	SubjectTemplate string `validate:"required"`

	// ConsumerPrefix is the prefix for the durable consumer name.
	//
	// The final durable name is constructed dynamically for each assigned partition:
	// "<ConsumerPrefix>_<partitionID>_<hash>".
	//
	// This ensures unique, stable durability for each partition assignment.
	ConsumerPrefix string `validate:"required"`

	// ProcessingGate configures optional exclusive processing enforcement.
	//
	// When enabled, the WorkerConsumer uses a distributed lock (via KV) to ensure
	// that it is the *only* active processor for its assigned partitions.
	// This prevents split-brain processing during rebalances.
	ProcessingGate *subscription.ProcessingGateConfig

	// Resolver configures the ownership resolver used when ProcessingGate is enabled.
	//
	// It defines how ownership is claimed, refreshed, and verified.
	Resolver subscription.ResolverConfig

	// PullGatingEnabled enables pre-pull ownership/state gating for consumers.
	//
	// When true, the consumer will check if it still owns the partition before
	// issuing a pull request to JetStream. This reduces "ghost" processing of
	// messages after assignment revocation.
	PullGatingEnabled bool

	// DrainOnRemove enables graceful draining when a partition assignment is revoked.
	//
	// When true, the consumer will stop pulling new messages but finish processing
	// buffered messages before shutting down the partition consumer.
	DrainOnRemove bool

	// DrainOnRemoveTimeout caps the time spent draining a revoked partition.
	//
	// If draining takes longer than this timeout, the consumer is forcibly closed.
	// Default: 10s.
	DrainOnRemoveTimeout time.Duration `default:"10s" validate:"gte=0"`

	// MaxConcurrentSubjects limits the number of partitions (subjects) processed concurrently.
	//
	// If the manager assigns more partitions than this limit, excess partitions
	// will be ignored (and logged/warned).
	MaxConcurrentSubjects int `validate:"gte=0"`

	// AllowWorkerIDChange controls whether the worker's identity can change during runtime.
	//
	// Default: false (immutable once set). Changing WorkerID usually requires a restart.
	AllowWorkerIDChange bool

	// Retry configures the backoff behavior for control-plane operations
	// (e.g., initial connection, creating consumers).
	Retry RetryConfig

	// IteratorEscalationWindow defines the sliding time window used to aggregate
	// iterator failures for escalation detection.
	//
	// If too many iterator errors occur within this window, the consumer will
	// attempt to escalate recovery.
	//
	// Default: 60s.
	IteratorEscalationWindow time.Duration `default:"60s" validate:"gt=0"`

	// IteratorEscalationThreshold is the number of iterator failures within the
	// escalation window that triggers consumer refresh/escalation.
	//
	// Default: 3.
	IteratorEscalationThreshold int `default:"3" validate:"gt=0"`

	// PartitionRefreshMinInterval sets the minimum interval between forced claim refreshes
	// per partition when pull gating is enabled.
	//
	// This prevents excessive load on the coordination backend (KV) during high-throughput pulling.
	//
	// Default: 500ms.
	PartitionRefreshMinInterval time.Duration `default:"500ms" validate:"gt=0"`

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)
}

DynamicConfig configures a Dynamic consumer. Uses unified naming; converted to subscription.WorkerConsumerConfig internally.

func (*DynamicConfig) SetDefaults 

func (c *DynamicConfig) SetDefaults() error

SetDefaults applies default values to the configuration.

func (*DynamicConfig) Validate 

func (c *DynamicConfig) Validate() error

Validate checks configuration constraints.

type DynamicOption 

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

DynamicOption applies only to Dynamic consumers.

This interface allows both universal Options and Dynamic-specific options (like WithProcessingGate) to be passed to NewDynamic.

func WithAllowWorkerIDChange 

func WithAllowWorkerIDChange(enabled bool) DynamicOption

WithAllowWorkerIDChange enables worker ID mutability (advanced).

func WithDrainOnRemove 

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

WithDrainOnRemove configures drain behavior on subject removal.

When enabled, revoked partitions will finish processing buffered messages before shutting down. The timeout caps the drain duration.

Parameters:

  • enabled: Whether to enable graceful draining
  • timeout: Maximum time to wait for draining (ignored if <= 0)

func WithIteratorEscalation 

func WithIteratorEscalation(window time.Duration, threshold int) DynamicOption

WithIteratorEscalation configures iterator failure escalation.

Only supported by Dynamic consumers. The escalation mechanism uses a sliding window to detect bursts of iterator failures and triggers consumer recreation when the threshold is exceeded.

func WithMaxConcurrentSubjects 

func WithMaxConcurrentSubjects(n int) DynamicOption

WithMaxConcurrentSubjects caps concurrent per-subject consumers.

func WithPartitionRefreshMinInterval 

func WithPartitionRefreshMinInterval(d time.Duration) DynamicOption

WithPartitionRefreshMinInterval sets the min interval for partition refresh.

func WithProcessingGate 

func WithProcessingGate(cfg *subscription.ProcessingGateConfig) DynamicOption

WithProcessingGate enables processing gate with given config.

func WithPullGating 

func WithPullGating(enabled bool) DynamicOption

WithPullGating enables pre-pull ownership checks.

func WithResolver 

func WithResolver(cfg subscription.ResolverConfig) DynamicOption

WithResolver configures the ownership resolver.

type MessageHandler 

type MessageHandler interface {
	// Handle processes a JetStream message.
	// Return nil to indicate successful processing; return an error for failures.
	// Ack/Nak behavior depends on the consumer's ManualAck setting:
	//   - ManualAck=false: nil → Ack, error → Nak (default behavior)
	//   - ManualAck=true: handler must call msg.Ack/Nak/Term explicitly
	Handle(ctx context.Context, msg jetstream.Msg) error
}

MessageHandler processes JetStream messages. This is the unified handler interface for all consumer types in this package.

func NewWIPHandler 

func NewWIPHandler(handler MessageHandler, cfg WIPConfig) MessageHandler

NewWIPHandler creates a work-in-progress handler wrapper.

The wrapper sends periodic msg.InProgress() calls while the underlying handler is processing, preventing JetStream from redelivering the message before processing completes.

Parameters:

  • handler: The underlying handler to wrap (nil returns nil)
  • cfg: Heartbeat configuration (interval, optional min interval, and logger)

Returns:

  • MessageHandler: The wrapped handler, or the original handler when disabled

Behavior:

  • cfg.Interval <= 0: Returns handler unchanged (heartbeats disabled)
  • cfg.Interval < MinInterval: Interval is clamped to MinInterval
  • handler == nil: Returns nil

Example: 

wrapped := consumer.NewWIPHandler(myHandler, consumer.WIPConfig{
    Interval: 10 * time.Second, // AckWait is 30s, so 10s is safe
    Logger:   logger,
})

type MessageHandlerFunc 

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

MessageHandlerFunc adapts a function to MessageHandler.

func (MessageHandlerFunc) Handle 

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

Handle implements MessageHandler.

type Option 

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

Option is a functional option that applies to all consumer types.

Common options (like WithLogger, WithAckWait) return this interface, allow them to be used with any consumer constructor (NewQueue, NewBroadcast, etc.).

func WithAckPolicy 

func WithAckPolicy(p jetstream.AckPolicy) Option

WithAckPolicy sets the JetStream ack policy.

func WithAckWait 

func WithAckWait(d time.Duration) Option

WithAckWait sets the time allowed for processing before redelivery.

If a message is not acknowledged within this duration, the server will redeliver it. Should be longer than expected processing time. Values <= 0 are ignored and the default (30s) is retained.

Parameters:

  • d: Acknowledgement wait duration (must be > 0)

func WithBatchSize 

func WithBatchSize(n int) Option

WithBatchSize sets the maximum number of messages to pull per request.

Higher batch sizes improve throughput but increase memory usage. Values <= 0 are ignored and the default (1) is retained.

Parameters:

  • n: Batch size (must be > 0)

func WithFetchTimeout 

func WithFetchTimeout(d time.Duration) Option

WithFetchTimeout sets the max time to wait when pulling a batch.

func WithInactiveThreshold 

func WithInactiveThreshold(d time.Duration) Option

WithInactiveThreshold sets how long an idle consumer is kept before cleanup.

func WithLogger 

func WithLogger(l types.Logger) Option

WithLogger sets the logger for consumer operations.

If nil is passed, the default no-op logger is retained.

Parameters:

  • l: Logger implementation (nil is ignored)

func WithManualAck 

func WithManualAck(enabled bool) Option

WithManualAck enables manual acknowledgement.

func WithMaxAckPending 

func WithMaxAckPending(n int) Option

WithMaxAckPending limits in-flight unacknowledged messages.

func WithMaxDeliver 

func WithMaxDeliver(n int) Option

WithMaxDeliver sets the maximum redelivery attempts.

func WithMaxWaiting 

func WithMaxWaiting(n int) Option

WithMaxWaiting caps outstanding pull requests.

func WithMetrics 

func WithMetrics(m types.MetricsCollector) Option

WithMetrics sets the metrics collector for consumer operations.

If nil is passed, the default no-op collector is retained.

Parameters:

  • m: Metrics collector implementation (nil is ignored)

type Queue 

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

Queue is a load-balanced consumer where multiple instances share one durable. Each message is delivered to exactly one instance (queue group semantics).

Unlike Broadcast (fan-out to all), Queue distributes messages across replicas. This is useful for classic worker queue patterns where each message should be processed by exactly one worker.

Lifecycle

Create with NewQueue, start consumption with Queue.Start, and clean up with Queue.Stop: 

q, err := consumer.NewQueue(js, "stream", "consumer", "subject.>", handler)
if err != nil { log.Fatal(err) }
defer q.Stop(ctx)

if err := q.Start(ctx); err != nil { log.Fatal(err) }

Thread Safety

Queue is safe for concurrent use. Queue.Start and Queue.Stop are serialized internally.

func NewQueue 

func NewQueue(
	js jetstream.JetStream,
	streamName, consumerName, filterSubject string,
	handler MessageHandler,
	opts ...QueueOption,
) (*Queue, error)

NewQueue creates a new queue (load-balanced) consumer.

func (*Queue) Start 

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

Start begins consuming messages.

func (*Queue) Stop 

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

Stop gracefully stops the consumer.

type QueueConfig 

type QueueConfig struct {
	CommonConfig

	// StreamName is the name of the JetStream stream to consume from.
	// This field is required and must match an existing stream.
	StreamName string `validate:"required"`

	// FilterSubject is the subject filter to consume from.
	// Supports wildcards (e.g., "orders.*", "events.>").
	// Only messages matching this filter will be delivered to the consumer.
	FilterSubject string `validate:"required"`

	// ConsumerName is the durable consumer name.
	// This name must be unique within the stream.
	// For Queue consumers, this name identifies the shared consumer group;
	// multiple instances using the same ConsumerName will share the load.
	ConsumerName string `validate:"required"`

	// Retry configures the backoff behavior for control-plane operations
	// (e.g., initial connection, creating the consumer).
	Retry RetryConfig

	// IteratorFactory optionally overrides the internal iterator creation logic.
	// This is primarily used for testing to inject mock iterators.
	IteratorFactory func(cons jetstream.Consumer, batch int, expiry time.Duration) (jetstream.MessagesContext, error)
}

QueueConfig configures a Queue consumer. Embeds CommonConfig for shared fields.

func DefaultQueueConfig 

func DefaultQueueConfig() QueueConfig

DefaultQueueConfig returns a QueueConfig with sensible defaults. Note: Required fields (StreamName, ConsumerName, FilterSubject) must still be set by the user.

func (*QueueConfig) SetDefaults 

func (c *QueueConfig) SetDefaults() error

SetDefaults sets default values for the configuration.

func (*QueueConfig) Validate 

func (c *QueueConfig) Validate() error

Validate checks configuration constraints.

type QueueOption 

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

QueueOption applies only to Queue consumers.

This interface allows both universal Options and Queue-specific options to be passed to NewQueue. Broadcast/Static/Dynamic specific options do not implement this, enforcing type safety.

type RetryConfig 

type RetryConfig struct {
	// Backoff is the delay between retries for control-plane operations.
	// Default: 100ms.
	Backoff time.Duration `default:"100ms" validate:"gte=0"`

	// Max caps the jittered backoff.
	// Default: 5s.
	Max time.Duration `default:"5s" validate:"gte=0,gtefield=Backoff"`

	// Multiplier grows the backoff window for decorrelated jitter.
	// Default: 1.6.
	Multiplier float64 `default:"1.6" validate:"gte=1"`

	// Base is the base backoff used for decorrelated jitter retries.
	// Default: 200ms. If zero and Backoff is set, Base falls back to Backoff.
	Base time.Duration `default:"200ms" validate:"gte=0"`

	// Seed optionally seeds the jitter RNG for deterministic tests.
	// When zero, a random seed is used.
	Seed int64
}

RetryConfig groups retry backoff settings.

type Static 

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

Static is a consumer bound to a single, fixed partition. Use for StatefulSet deployments where pod ordinal determines partition.

Lifecycle

Create with NewStatic, start consumption with Static.Start, and clean up with Static.Stop: 

consumer, err := consumer.NewStatic(js, "stream", "consumer-0", "events.{{partition}}", 10, 0, handler)
if err != nil { log.Fatal(err) }
defer consumer.Stop(ctx)

if err := consumer.Start(ctx); err != nil { log.Fatal(err) }

Thread Safety

Static is safe for concurrent use. Static.Start and Static.Stop are serialized internally.

Deprecation Notice

This type wraps partition.JSConsumer. Future versions may deprecate the partition package in favor of this unified consumer API.

func NewStatic 

func NewStatic(
	js jetstream.JetStream,
	streamName, consumerName, subjectPattern string,
	numPartitions, partIdx int,
	handler MessageHandler,
	opts ...StaticOption,
) (*Static, error)

NewStatic creates a new static partition consumer.

func (*Static) Partition 

func (s *Static) Partition() int

Partition returns the partition index this consumer handles.

Returns:

  • int: The zero-based partition index (0 to NumPartitions-1).

func (*Static) Start 

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

Start begins consuming messages in a background goroutine.

The consumer creates or binds to a durable JetStream consumer and starts a pull loop. Messages are delivered to the handler configured at creation.

Start may only be called once. Calling Start on an already-started consumer returns an error.

Parameters:

  • ctx: Context for lifecycle control. Cancellation stops the consumer.

Returns:

  • error: Non-nil if the consumer is already started or if JetStream consumer creation fails.

func (*Static) Stop 

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

Stop gracefully stops the consumer.

Stop cancels the internal pull loop and waits for pending message processing to complete (up to the context deadline). The underlying JetStream consumer is NOT deleted; it will be garbage-collected by the server after InactiveThreshold.

Stop is idempotent; calling it multiple times is safe.

Parameters:

  • ctx: Context with shutdown deadline. If the deadline expires, Stop returns context.DeadlineExceeded but the consumer will still eventually stop.

Returns:

  • error: Context error if the wait times out; nil otherwise.

func (*Static) Subject 

func (s *Static) Subject() string

Subject returns the NATS subject this consumer subscribes to.

The subject is derived from the SubjectPattern with the partition placeholder replaced by the actual partition index. If the pattern contains {{key}}, it is replaced with a wildcard (*) for subscription.

Returns:

  • string: The filter subject, e.g., "events.*.0" or "orders.2.>".

type StaticConfig 

type StaticConfig struct {
	CommonConfig

	// StreamName is the JetStream stream to consume from.
	// Required.
	StreamName string `validate:"required"`

	// NumPartitions is the total number of partitions for the stream.
	//
	// This defines the sharding factor. Messages are consistently hashed
	// to a partition index in the range [0, NumPartitions-1].
	// Must be > 0.
	//
	// WARNING: Changing NumPartitions changes the hash mapping.
	NumPartitions int `validate:"required,gt=0"`

	// Partition is the specific partition index this consumer should process.
	//
	// Must be in the range [0, NumPartitions-1].
	// Typically assigned based on the application instance's ordinal (e.g., StatefulSet index).
	Partition int `validate:"gte=0"`

	// SubjectPattern is the subject template with placeholders.
	//
	// Placeholders:
	//   - {{partition}} - Replaced with partition index (0 to N-1). Required.
	//   - {{key}}       - Replaced with the partition key. Optional.
	//
	// Placeholders must occupy a full token between dots. Embedded placeholders
	// like "events.{{partition}}-v1" are invalid.
	//
	// Examples:
	//   - "events.completed.{{partition}}"       → "events.completed.0"
	//   - "events.{{key}}.{{partition}}"         → "events.tool-abc.3"
	//   - "orders.{{partition}}.{{key}}.created" → "orders.2.customer-xyz.created"
	//
	// Validation:
	//   - Must contain {{partition}} placeholder
	//   - Must not produce empty NATS subject tokens (e.g., "events..{{partition}}" is invalid)
	SubjectPattern string `validate:"required"`

	// ConsumerName is the durable consumer name.
	//
	// Must be unique per partition.
	// The final durable name on the server often incorporates the partition index
	// to avoid collisions between partitions, or the user must ensure uniqueness.
	ConsumerName string `validate:"required"`

	// HashSeed is an optional seed for the consistent hashing algorithm.
	//
	// Using a consistent seed ensures that the same message key always maps to
	// the same partition index across restarts/redeployments.
	HashSeed uint64
}

StaticConfig configures a Static consumer. Uses unified naming; converted to partition.ConsumerConfig internally.

func (*StaticConfig) SetDefaults 

func (c *StaticConfig) SetDefaults() error

SetDefaults applies default values to the configuration.

func (*StaticConfig) Validate 

func (c *StaticConfig) Validate() error

Validate checks configuration constraints.

type StaticOption 

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

StaticOption applies only to Static consumers.

This interface allows both universal Options and Static-specific options (like WithHashSeed) to be passed to NewStatic.

func WithDispatchByKey 

func WithDispatchByKey() StaticOption

WithDispatchByKey enables per-key concurrent message processing.

When enabled, messages are routed to separate goroutines based on their key. Messages with the same key are processed sequentially (preserving order), while different keys are processed concurrently in parallel goroutines.

IMPORTANT: SubjectPattern MUST contain {{key}} placeholder when DispatchByKey is enabled. The key is extracted based on the {{key}} position in the pattern. For example, with pattern "events.{{partition}}.{{key}}" and subject "events.0.customer-abc", the key is "customer-abc".

Placeholders must occupy a full token between dots. Embedded placeholders like "events.{{key}}-v1.{{partition}}" are invalid.

WARNING: This creates an UNBOUNDED number of goroutines - one goroutine per unique key. If your workload has millions of unique keys, memory usage will grow proportionally. Goroutines are cleaned up after KeyIdleTimeout of inactivity.

Use this when:

  • You need per-key ordering but want parallelism across keys
  • Your key cardinality is bounded (e.g., thousands, not millions)
  • Slow processing of one key should not block other keys

func WithHashSeed 

func WithHashSeed(seed uint64) StaticOption

WithHashSeed sets the consistent hashing seed.

func WithKeyChannelBuffer 

func WithKeyChannelBuffer(size int) StaticOption

WithKeyChannelBuffer sets the buffer size for each key's message channel.

When the buffer is full, the main pull loop blocks (backpressure). Larger buffers absorb bursts but use more memory per active key.

Only used when DispatchByKey is enabled. Default: 32

func WithKeyExtractor 

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

WithKeyExtractor sets a custom key extraction function.

The extracted key determines which goroutine processes the message. Messages with the same key are guaranteed to be processed sequentially.

If not set, uses a pattern-aware extractor based on the {{key}} position in SubjectPattern. For example, with pattern "events.{{partition}}.{{key}}" and subject "events.0.customer-abc", the key is "customer-abc".

Only used when DispatchByKey is enabled.

func WithKeyIdleTimeout 

func WithKeyIdleTimeout(d time.Duration) StaticOption

WithKeyIdleTimeout sets how long an idle key goroutine waits before exiting.

After this duration with no messages, the goroutine exits and is removed. A new goroutine is created if messages for that key arrive later.

Only used when DispatchByKey is enabled. Default: 30s

type WIPConfig 

type WIPConfig struct {
	// Interval is the heartbeat interval for msg.InProgress() calls.
	//
	// Recommended: set Interval to AckWait/3 for safety margin.
	// Values <= 0 disable heartbeats (returns unwrapped handler).
	// Values < DefaultWIPMinInterval are clamped to prevent overhead.
	Interval time.Duration

	// MinInterval overrides the minimum interval threshold.
	// If zero, DefaultWIPMinInterval (100ms) is used.
	// Set to a negative value to disable clamping (not recommended).
	MinInterval time.Duration

	// Logger receives heartbeat errors. If nil, errors are silently ignored.
	Logger types.Logger
}

WIPConfig configures the work-in-progress heartbeat wrapper.

The wrapper periodically calls msg.InProgress() while the handler is running to extend the JetStream AckWait deadline for long-running processing.

Interval Selection Guidelines

Choose Interval based on your AckWait setting:

  • Recommended: AckWait / 3 (provides safety margin)
  • Maximum safe: AckWait / 2 (minimum margin)
  • Example: AckWait=30s → Interval=10s

Very small intervals (< 100ms) are clamped to DefaultWIPMinInterval to prevent excessive msg.InProgress() calls which can degrade performance.

Performance Characteristics

The wrapper uses lazy initialization:

  • Fast handlers (< Interval): Only a timer allocation, no goroutine spawned
  • Slow handlers (>= Interval): One goroutine per message during processing

Compatibility

WIPHandler works with all consumer types (Queue, Static, Dynamic, Broadcast) and both auto-ack and manual-ack modes. In manual-ack mode, the handler must still call msg.Ack/Nak/Term exactly once; the wrapper only calls InProgress().

type WIPHandler 

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

WIPHandler wraps a MessageHandler with automatic msg.InProgress() heartbeats.

The heartbeat starts lazily: if the handler finishes before Interval elapses, no heartbeat goroutine is started. This ensures minimal overhead for fast handlers.

Thread Safety: WIPHandler is safe for concurrent use. Each Handle call operates independently with its own heartbeat goroutine (if needed).

func (*WIPHandler) Handle 

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

Handle processes a message while extending its AckWait deadline with heartbeats.

Parameters:

  • ctx: Request-scoped context for cancellation and deadlines
  • msg: The JetStream message to process

Returns:

  • error: Any error returned by the wrapped handler

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.