consumer

Documentation

Overview

Package consumer provides the public API for creating and managing RedisSMQ consumers.

A consumer subscribes to one or more queues and processes messages using user-defined handlers. It manages heartbeats, background workers, and graceful shutdown automatically.

Index

• type Consumer
  • func New(opts ...c.Option) *Consumer
  • func (cons *Consumer) Cancel(queue *q.QueueParams) *Consumer
  • func (cons *Consumer) CancelWithGroup(queue *q.QueueParams, groupID string) *Consumer
  • func (cons *Consumer) Consume(queue *q.QueueParams, handler Handler) *Consumer
  • func (cons *Consumer) ConsumeWithGroup(queue *q.QueueParams, groupID string, handler Handler) *Consumer
  • func (cons *Consumer) ID() string
  • func (cons *Consumer) IsRunning() bool
  • func (cons *Consumer) Queues() []*q.QueueParams
  • func (cons *Consumer) Run(ctx context.Context) error
  • func (cons *Consumer) Shutdown()
• type Handler

Types

type Consumer

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

Consumer is a message consumer.

It receives messages from one or more queues and invokes user-supplied handlers. A consumer must be started with Run before it begins processing. After use, it should be stopped with Shutdown.

func New

func New(opts ...c.Option) *Consumer

New creates a new consumer instance with the provided options.

By default, the consumer uses a 60-second heartbeat TTL and disabled batch acknowledgments/unacknowledgments. Use options like c.WithHeartbeatTTL or c.WithBatchAcks to customise behavior.

func (*Consumer) Cancel

func (cons *Consumer) Cancel(queue *q.QueueParams) *Consumer

Cancel removes a previously registered message handler from a queue.

func (*Consumer) CancelWithGroup

func (cons *Consumer) CancelWithGroup(queue *q.QueueParams, groupID string) *Consumer

CancelWithGroup removes a previously registered message handler from a Pub/Sub consumer group.

func (*Consumer) Consume

func (cons *Consumer) Consume(queue *q.QueueParams, handler Handler) *Consumer

Consume registers a message handler for a queue.

The handler is called for each message dequeued from the queue. If the handler returns an error, the message is unacknowledged and may be retried or dead-lettered according to its retry policy.

func (*Consumer) ConsumeWithGroup

func (cons *Consumer) ConsumeWithGroup(queue *q.QueueParams, groupID string, handler Handler) *Consumer

ConsumeWithGroup registers a message handler for a Pub/Sub queue and consumer group.

The handler is called for messages delivered to the specified consumer group. If the queue is not Pub/Sub, an error is returned when the consumer starts.

func (*Consumer) ID

func (cons *Consumer) ID() string

ID returns the unique identifier of the consumer.

func (*Consumer) IsRunning

func (cons *Consumer) IsRunning() bool

IsRunning reports whether the consumer is currently running.

func (*Consumer) Queues

func (cons *Consumer) Queues() []*q.QueueParams

Queues returns the queue parameters for all registered handlers.

func (*Consumer) Run

func (cons *Consumer) Run(ctx context.Context) error

Run starts the consumer and all registered message handlers.

It is safe to call multiple times. If no handlers are registered, it returns c.ErrNoQueues.

func (*Consumer) Shutdown

func (cons *Consumer) Shutdown()

Shutdown gracefully stops the consumer.

It returns in-flight messages to the pending queue, stops heartbeats, and removes the consumer from all subscribed queues. Shutdown is idempotent and safe to call multiple times.

type Handler

type Handler = internalConsumer.Handler

Handler is the function signature for processing a message.

It is an alias of the internal consumer handler type so that external users do not need to import internal packages.