class LavinMQ::AMQP::StreamQueue::StreamQueueMessageStore

Defined in:

lavinmq/amqp/queue/stream_queue_message_store.cr

Constructors

Instance Method Summary

Instance methods inherited from class LavinMQ::Queue::MessageStore

[](sp : SegmentPosition) : BytesMessage [], avg_bytesize : UInt32 avg_bytesize, bytesize : UInt64 bytesize, close : Nil close, closed : Bool closed, delete(sp) : Nil
delete delete, delete_file(file) delete_file, empty? empty?, empty_change : Channel(Bool) empty_change, first? : Envelope | Nil first?, purge(max_count : Int = UInt32::MAX) : UInt32 purge, push(msg) : SegmentPosition push, requeue(sp : SegmentPosition) requeue, shift?(consumer = nil) : Envelope | Nil shift?, size : UInt32 size, unmap_segments(except : Enumerable(UInt32) = StaticArray(UInt32, 0).new(0_u32)) unmap_segments

Constructor methods inherited from class LavinMQ::Queue::MessageStore

new(queue_data_dir : String, replicator : Clustering::Replicator | Nil, durable : Bool = true, metadata : ::Log::Metadata = ::Log::Metadata.empty) new

Constructor Detail

def self.new(*args, **kwargs) #

[View source]

Instance Method Detail

def cleanup_consumer_offsets #

[View source]
def consumer_offset_file_full?(consumer_tag) #

[View source]
def delete(sp) : Nil #

[View source]
def drop_overflow #

[View source]
def 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


[View source]
def last_offset : Int64 #

[View source]
def last_offset_by_consumer_tag(consumer_tag) #

[View source]
def max_age : Time::Span | Time::MonthSpan | Nil #

[View source]
def max_age=(max_age : Time::Span | Time::MonthSpan | Nil) #

[View source]
def max_length : Int64 | Nil #

[View source]
def max_length=(max_length : Int64 | Nil) #

[View source]
def max_length_bytes : Int64 | Nil #

[View source]
def max_length_bytes=(max_length_bytes : Int64 | Nil) #

[View source]
def new_messages : Channel(Bool) #

[View source]
def purge(max_count : Int = UInt32::MAX) : UInt32 #
Description copied from class LavinMQ::Queue::MessageStore

Deletes all "ready" messages (not unacked)


[View source]
def push(msg) : SegmentPosition #

[View source]
def replace_offsets_file(capacity : Int, &) #

[View source]
def shift?(consumer : AMQP::StreamConsumer) : Envelope | Nil #

[View source]
def store_consumer_offset(consumer_tag : String, new_offset : Int64) #

[View source]