class

LavinMQ::AMQP::Stream

Inherits LavinMQ::AMQP::DurableQueue < LavinMQ::AMQP::Queue < LavinMQ::AMQP::Argument::DeadLettering < LavinMQ::AMQP::ArgumentTarget < LavinMQ::AMQP::QueueStats < LavinMQ::Stats < LavinMQ::SortableJSON < LavinMQ::PolicyTarget < Reference < Object

Constants

INVALID_ARGUMENTS = {"x-dead-letter-exchange", "x-dead-letter-routing-key", "x-expires", "x-delivery-limit", "x-delayed-retry-min", "x-delayed-retry-max", "x-delayed-retry-multiplier", "x-overflow", "x-single-active-consumer", "x-max-priority"}

Arguments that have no meaning for streams. Rejected at queue declare AND when matched by a policy — otherwise a policy can quietly install state (e.g. delivery-limit spawning the inherited drop_redelivered fiber, which dereferences the legacy @rfile that streams don't maintain) and crash the server.

Class methods

create(vhost : VHost, name : String, exclusive : Bool = false, auto_delete : Bool = false, arguments : AMQP::Table = AMQP::Table.new)
Source
validate_arguments!(arguments)
Source

Instance methods

ack(sp : SegmentPosition) : Nil
Source
add_consumer(consumer : Client::Channel::Consumer) : Bool

Adds the consumer unless the queue is in exclusive use, checked atomically with the add. Returns false if the consumer was refused for that reason, raises ClosedError if the queue is closed or about to be deleted.

Source
basic_get(no_ack, force = false, & : Envelope -> Nil) : Bool

Streams does not support basic_get, so always returns false

Source
consume_get(consumer : AMQP::StreamConsumer, & : Envelope -> Nil) : Bool
Source
find_offset(offset, tag = nil, track_offset = false) : Tuple(Int64, UInt32, UInt32)
Source
last_offset(*args, **options)
Source
last_offset(*args, **options, &)
Source
new_messages(*args, **options)
Source
new_messages(*args, **options, &)
Source
publish(msg : Message) : PublishResult
Source
purge(max_count : Int = UInt32::MAX) : UInt32
Source
push_requeued(consumer : AMQP::StreamConsumer, sp : SegmentPosition) : Bool

The deliver path shifts from the consumer's requeued messages under @msg_store_lock (see StreamMessageStore#shift?). Returns whether there were no requeued messages before.

Source
reader(offset)
Source
reject(sp : SegmentPosition, requeue : Bool)

Consumer-initiated reject (basic.reject and basic.nack): delays via the retry queue if configured

Source
requeue(sp : SegmentPosition)

Broker-initiated requeue (channel close, basic.recover): always instant, never via the retry queue

Source
rm_consumer(consumer : Client::Channel::Consumer)
Source
store_consumer_offset(consumer_tag : String, offset : Int64) : Nil
Source
stream_msg_store
Source