class

Crabbit::ConsumerOptions

Inherits Reference < Object

Configures subscription position, credit, processing, offset storage, filtering, Single Active Consumer behavior, and recovery.

Constructors

new(name : String | Nil = nil, offset : OffsetSpecification = OffsetSpecification.next, initial_credit : UInt16 = 10_u16, filters : Array(String) = [] of String, match_unfiltered : Bool = false, single_active_consumer : Bool = false, super_stream : String | Nil = nil, concurrency : Int32 = 1, buffer_size : Int32 = 1024, validate_crc : Bool = true, auto_store_every : Int32 | Nil = nil, auto_store_interval : Time::Span | Nil = nil, recovery_policy : RecoveryPolicy = RecoveryPolicy.new, on_consumer_update : Proc(Bool, OffsetSpecification) | Nil = nil, on_consumer_update_context : Proc(ConsumerUpdateContext, OffsetSpecification) | Nil = nil, subscription_offset : Proc(String, OffsetSpecification) | Nil = nil, on_state_change : Proc(ResourceEvent, Nil) | Nil = nil, topology_refresh : Time::Span = 30.seconds)

Creates consumer options.

initial_credit is measured in chunks, while buffer_size is measured in logical messages. Callback consumers use concurrency handler fibers; pull consumers ignore it. Broker filtering accepts multiple OR-matched filters and can optionally include unfiltered messages.

Automatic storage requires name. auto_store_every stores after a number of processed deliveries, and auto_store_interval stores the latest contiguous processed offset periodically. With Single Active Consumer enabled, one of the update callbacks may choose the offset each time the subscription becomes active or inactive.

Source

Instance methods

auto_store_every

Returns the optional processed-message threshold for automatic offset storage.

Source
auto_store_interval

Returns the optional interval for automatic offset storage.

Source
buffer_size

Returns the capacity of the logical delivery queue.

Source
concurrency

Returns the number of callback-processing fibers.

Source
filters

Returns the server-side filter values.

Source
initial_credit

Returns the initial number of broker chunks that may be in flight.

Source
match_unfiltered

Returns whether messages without a filter value also match.

Source
name

Returns the optional consumer reference used for stored offsets and SAC.

Source
offset

Returns the initial subscription position.

Source
on_consumer_update

Returns the legacy Single Active Consumer update listener.

Source
on_consumer_update_context

Returns the context-aware Single Active Consumer update listener.

Source
on_state_change

Returns the optional asynchronous lifecycle listener.

Source
recovery_policy

Returns the reconnect backoff policy.

Source
single_active_consumer

Returns whether Single Active Consumer semantics are enabled.

Source
subscription_offset

Returns the optional per-partition starting-offset resolver.

Source
super_stream

Returns the parent super-stream name added to subscription properties.

Source
topology_refresh

Returns the super-stream partition refresh interval.

Source
validate_crc

Returns whether Deliver chunk CRC32 values are verified.

Source