Consume messages

Consume(ctx context.Context, consumerFunc ConsumerFunc[T], options *ConsumeOptions) error

Processes messages for a registered consumer group. Blocks until the session stops. Cancel ctx to begin graceful shutdown.

Example

Run the charge-cards group on the payments.requested stream using an initialized client and the example payload. This consumer handler prints each payment’s order id.

func run(client *sqlstreams.Client) error {
  ctx, stop := sqlstreams.LifecycleContext(nil)
  defer stop()

  payments := client.Stream[PaymentRequestedV1]("payments.requested")
  chargeCards := payments.Consumer("charge-cards")
  consumer, err := chargeCards.Register(ctx, nil)
  if err != nil {
    return err
  }
  return consumer.Consume(ctx, printPayment, &sqlstreams.ConsumeOptions{
    BatchLimit:         4,
    QueueSize:          4,
    MessageConcurrency: 1,
  })
}

func printPayment(ctx context.Context, payment *PaymentRequestedV1) error {
  fmt.Println(payment.OrderId)
  return nil
}

Parameters

ConsumeOptions

These options apply to one Consume call and can differ across instances. They are not stored in the group’s declaration. Changing the original struct does not reconfigure a running session.

nil options and zero fields select defaults, except where noted. Negative counts and durations are invalid.

ShutdownTimeout

For the charge-cards group with a 30s message timeout ceiling and default margins, the derived budget is 30s + 100ms + 2s = 32.1s. A refreshed group ceiling changes this derived budget.

QueueMargin is excluded because shutdown waits for in-flight processing and recording, rather than allowing more messages to wait in the queue.

Returns

Message results

The return value of consumerFunc determines the message’s outcome. A message-processing error does not ordinarily become Consume’s returned error.

A DelayedDelivery error takes precedence when classifying a wrapped error. The group’s retry policy limits failed attempts and consumer handler-requested delays. Message metadata provides the current counts and resolved options inside the consumer handler.

Behavior

Instances and bindings

Consume reads the payload version declared by T. Instances using the same group share progress. The call can wait for an incompatible live binding declaration before it begins processing messages.

Only one Consume call can run on an instance at a time. A new call is allowed after the previous call finishes.

System manager

With a cancellable context, Consume also runs the system manager unless ClientConfig.DisableManager is set. A manager setup error stops the paired session. Errors from its running loop are logged and retried. The session’s exit stops its paired manager.

Shutdown

Cancellation stops new work and gives in-flight processing and recording up to ShutdownTimeout to finish. Stop a consumer gracefully shows how to handle cancellation in the application.

Inspecting settings

The Info-level starting log reports resolved queue, concurrency, polling, and timing settings. Enable Info in the application’s logger to see it. The printed shutdown timeout is calculated at startup and can later change when the timeout is derived. The consumer worker list command shows stored group configuration, which excludes these per-call options.

Errors

Related topics

• Size a consumer queue — Tune batching, queue size, and concurrency.
• Stop a consumer gracefully — Connect signals and let active consumer handlers finish.
• Make message processing safe to repeat — Prevent duplicate application effects.
• Consumer leases — How queue waiting, timeouts, and reclaim fit together.