class

LavinMQ::AMQP::ConsumerOffsets

Owns a stream's consumer_offsets file: an append-only log of {consumer_tag, offset} entries used to resume consumers where they left off. The file is compacted (via cleanup) to drop offsets that have fallen out of the stream and to keep the file size bounded.

Constants

ENTRY_OVERHEAD = 1 + 8

Log = LavinMQ::Log.for("consumer_offsets")

MAX_ENTRY_SIZE = 255 + ENTRY_OVERHEAD

MAX_FILE_SIZE = (64_i64 * 1024) * 1024

Size bounds for the compacted consumer_offsets file.

MIN_FILE_SIZE = 8_i64 * 1024

Constructors

new(dir : String, capacity : Int, replicator : Clustering::Replicator | Nil)

Source

Class methods

entry_size(consumer_tag : String) : Int64

Encoded size of one {consumer_tag, offset} entry in the offsets file.

Source

trim_to_size(offsets : Array(Tuple(String, Int64, Int64)), max_size : Int64) : Array(Tuple(String, Int64, Int64))

Drops the oldest (lowest position) offsets until the set fits in max_size. Survivors are returned oldest first so rewriting them keeps file position reflecting commit recency. Usually drops nothing.

Source

Instance methods

cleanup(force = false, & : -> Int64)

Compacts the file, dropping offsets that are no longer in the stream and capping the file size. The block yields the lowest offset still in the stream; it is only evaluated when there is something to compact. force skips the unchanged-floor short-circuit, for callers (store) that need the file's capacity actually enforced, not just offsets dropped.

Source

delete

Source

last_offset_by_tag(consumer_tag)

Source

size

Current size of the offsets file on disk.

Source

store(consumer_tag : String, new_offset : Int64, & : -> Int64)

Appends new_offset for consumer_tag. Compacts the file first if the append wouldn't fit; the block yields the lowest offset still in the stream so stale offsets can be dropped during compaction.

Source