class

LavinMQ::AMQP::StreamMessageStore

Inherits LavinMQ::MessageStore < Reference < Object

Constructors

new(*args, **kwargs)
Source

Instance methods

acquire_segment(consumer : StreamConsumer) : Nil
Source
cleanup_consumer_offsets
Source
close
Source
delete(sp) : Nil
Source
delete
Source
drop_expired
Source
drop_overflow
Source
expiry_changed

Signalled when the next max-age expiry may have moved: a new segment was opened or max-age changed. Closed with the store.

Source
find_offset(offset, tag = nil, track_offset = false) : Tuple(Int64, UInt32, UInt32)

Used once when a consumer is started Populates segment and position by iterating through segments until offset is found ameba:disable Metrics/CyclomaticComplexity

Source
last_offset
Source
last_offset_by_consumer_tag(consumer_tag)
Source
max_age
Source
max_age=(max_age : Time::Span | Time::MonthSpan | Nil)
Source
max_length
Source
max_length=(max_length : Int64 | Nil)
Source
max_length_bytes
Source
max_length_bytes=(max_length_bytes : Int64 | Nil)
Source
new_messages
Source
next_expiry

When the oldest segment expires, nil if no segment can be dropped (no max-age, or only the write segment is left)

Source
next_segment_id(segment) : UInt32 | Nil
Source
next_segment_offset(segment) : Tuple(UInt32, Int64) | Nil

The segment after segment and the offset of its first message

Source
purge(max_count : Int = UInt32::MAX) : UInt32

Deletes all "ready" messages (not unacked)

Source
push(msg) : SegmentPosition
Source
read(segment : UInt32, position : UInt32) : Envelope | Nil
Source
read_with_lease?(lock : Mutex, segment : UInt32, position : UInt32, & : Envelope -> _) : Bool

Like #read, but yields the message outside lock, see #shift_with_lease?

Source
release_segment(consumer : StreamConsumer) : Nil
Source
shift?(consumer : AMQP::StreamConsumer) : Envelope | Nil
Source
store_consumer_offset(consumer_tag : String, new_offset : Int64)
Source
unmap_if_unused(seg : UInt32) : Nil

Drops a segment's pages from memory once no consumer is reading from it. The active write segment is kept mapped.

Source

Nested types