class Crabbit::Consumer

• Crabbit::Consumer
• Reference
• Object

Overview

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.

Included Modules

• Enumerable(Crabbit::Delivery)

Defined in:

crabbit/consumer.cr

Constructors

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

  Creates and subscribes a consumer.

Instance Method Summary

• #close : Nil

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

• #closed? : Bool

  Returns whether the consumer was permanently closed.

• #each(&block : Delivery -> ) : Nil

  Yields deliveries until the consumer closes.

• #open? : Bool

  Returns whether the consumer currently has an active subscription.

• #options : ConsumerOptions

  Returns the immutable consumer options.

• #receive : Delivery

  Receives one delivery, waiting while the queue is empty.

• #receive? : Delivery | Nil

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

• #state : ResourceState

  Returns the current lifecycle state.

• #store_offset(offset : UInt64) : Nil

  Stores an absolute offset for this named consumer.

• #store_offset(delivery : Delivery) : Nil

  Stores the offset of delivery for this named consumer.

• #stored_offset : UInt64 | Nil

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

• #stream : String

  Returns the subscribed stream name.

Constructor Detail

Instance Method Detail