class

JoobQ::RedisEventSink

Inherits JoobQ::EventSink < Reference < Object

Persists events in a capped Redis Stream (replayable by id) and publishes them on a pub/sub channel for live consumers. Delivery is batched: one pipelined round trip appends to the stream, a second publishes the batch.

Constants

DEFAULT_MAX_BYTES = (64_i64 * 1024) * 1024

LIVE = "joobq:events:live"

MAX_LEN = 50000

STREAM = "joobq:events"

TRIM_FLOOR = 100_i64

The stream is never trimmed below this many entries, whatever they weigh.

TRIM_LOCK = "joobq:events:trim_lock"

TRIM_LOCK_SECONDS = 10

Constructors

new(redis : Redis::PooledClient = RedisStore.instance.redis)

Source

Class methods

max_bytes

The most memory the stream may use, in bytes (JOOBQ_EVENTS_MAX_BYTES; 0 turns the limit off, as does a negative number). A value that is not a whole number is ignored with a warning.

Source

trim_to_bytes(redis : Redis::PooledClient = RedisStore.instance.redis, budget : Int64 = max_bytes) : Int64

Drops the oldest entries when the stream uses more memory than budget; returns how many it dropped. MAXLEN bounds the count only, and events with big payloads can make 50 000 entries very heavy. Only one process checks per lock period (the lock is a throttle, not held during the trim), and the stream is trimmed to 90 % of the budget so it does not trim on every beat.

Source

Instance methods

publish(event : Event) : Nil

Source

publish_all(events : Array(Event)) : Nil

Sinks that can deliver in bulk (e.g. one pipelined round trip) override this.

Source