class

Crabbit::Consumer

Inherits Enumerable < Reference < Object

Recovering pull or callback consumer for one RabbitMQ stream.

Pull mode uses #receive, #receive?, or #each. Callback mode is created by Environment#consumer(stream, options) { |delivery| ... } and marks each delivery processed after the handler returns. A consumer reconnects and resumes from the latest contiguous broker-delivery prefix until closed.

Constructors

new(environment : Environment, stream : String, options : ConsumerOptions, handler : Proc(Delivery, Nil) | Nil = nil)

Creates and subscribes a consumer.

Applications normally use the Environment#consumer overload matching pull or callback mode.

Source

Instance methods

close

Idempotently unsubscribes, stores any pending automatic offset, and closes the delivery queue.

Source
closed?

Returns whether the consumer was permanently closed.

Source
each

Yields deliveries until the consumer closes.

Enumerable consumption acknowledges after the block returns, including when it raises. Use #receive for explicit acknowledgement control.

Source
open?

Returns whether the consumer currently has an active subscription.

Source
options

Returns the immutable consumer options.

Source
receive

Receives one delivery, waiting while the queue is empty.

The caller owns acknowledgement and must invoke Delivery#processed! when processing is complete. Raises ResourceClosedError after close.

Source
receive?

Receives one delivery, waiting while the queue is empty, or returns nil after the consumer closes and the queue drains.

The caller must invoke Delivery#processed! for every returned delivery.

Source
state

Returns the current lifecycle state.

Source
store_offset(offset : UInt64) : Nil

Stores an absolute offset for this named consumer.

Raises ConfigurationError when ConsumerOptions#name is absent.

Source
store_offset(delivery : Delivery) : Nil

Stores the offset of delivery for this named consumer.

Raises ArgumentError when the delivery belongs to another stream.

Source
stored_offset

Returns this named consumer's stored offset, or nil when absent.

Source
stream

Returns the subscribed stream name.

Source