🥮 mooncakes.io
    Sign in

    relay

    Distributed message queue with V3 reliability and Valkey support

    Download zip
    Author
    Version
    0.3.0
    License
    Apache-2.0
    Repository
    Last updated
    4 months ago
    Downloads
    133

    Metalymph/relay does not have a README file

    QueueError

    source
    pub suberror QueueError {
      OperationFailed(String)
    } derive(ToJson, 
    Debug    )

    QueueError represents errors related to queue lifecycle and operations.

    Dashboard

    source
    pub struct Dashboard[T] {
      queue : RelayQueue[T]
      backend : InMemoryBackend[T]
    }

    Dashboard provides a visual representation of the RelayQueue state.

    Dashboard::new

    source
    fn[T] Dashboard::new(queue : RelayQueue[T], backend : InMemoryBackend[T]) -> Dashboard[T]

    new creates a new dashboard for the provided queue and backend.

    Dashboard::render

    source
    fn[T] Dashboard::render(self : Dashboard[T]) -> Unit

    render prints an ANSI color-coded snapshot of the current queue state.

    Dashboard::start_auto_render

    source
    fn[T] Dashboard::start_auto_render(self : Dashboard[T]) -> Unit

    start_auto_render starts an infinite loop of rendering.

    InMemoryBackend

    source
    pub struct InMemoryBackend[T] {
      queue : 
    Queue    [Message[T]]
      dlq : 
    Queue    [Message[T]]
      in_flight : 
    HashMap    [String, Message[T]]
      policy : RetryPolicy
      next_id : Int
      count : Int
      dlq_count : Int
    }

    InMemoryBackend uses @aqueue.Queue from moonbitlang/async for local, non-blocking storage.

    InMemoryBackend::ack

    source
    fn[T] InMemoryBackend::ack(self : InMemoryBackend[T], id : String) -> Unit

    ack confirms successful processing and removes from in-flight.

    InMemoryBackend::dlq_size

    source
    fn[T] InMemoryBackend::dlq_size(self : InMemoryBackend[T]) -> Int

    dlq_size returns the count of messages in the Dead Letter Queue.

    InMemoryBackend::nack

    source
    async fn[T] InMemoryBackend::nack(self : InMemoryBackend[T], id : String) -> Unit

    nack re-inserts the message into the queue or moves to DLQ if max attempts reached.

    InMemoryBackend::new

    source
    fn[T] InMemoryBackend::new(capacity : Int, policy? : RetryPolicy) -> InMemoryBackend[T]

    new creates a new InMemoryBackend with a fixed capacity and optional policy.

    InMemoryBackend::pop

    source
    async fn[T] InMemoryBackend::pop(self : InMemoryBackend[T]) -> Message[T]

    pop retrieves a message and marks it as in-flight.

    InMemoryBackend::push

    source
    async fn[T] InMemoryBackend::push(self : InMemoryBackend[T], v : T) -> Unit

    push adds a message to the internal queue.

    InMemoryBackend::size

    source
    fn[T] InMemoryBackend::size(self : InMemoryBackend[T]) -> Int

    size returns the cached count of messages in the main queue.

    InMemoryBackend::to_relay_queue

    source
    fn[T] InMemoryBackend::to_relay_queue(self : InMemoryBackend[T]) -> RelayQueue[T]

    to_relay_queue converts the backend into a generic RelayQueue interface.

    Message

    source
    pub struct Message[T] {
      id : String
      payload : T
      attempt_count : Int
    }

    Message[T] represents a wrapper for data sent through the queue.

    Message::new

    source
    fn[T] Message::new(id : String, payload : T, attempt_count? : Int) -> Message[T]

    new creates a new message.

    RedisBackend

    source
    pub struct RedisBackend[T] {
      client : 
    Client
      key : String
      serialize : (T) -> String
      deserialize : (String) -> T
      next_id : Int
    }

    RedisBackend implements the storage interface for Relay using Redis/Valkey.

    RedisBackend::ack

    source
    async fn[T] RedisBackend::ack(_self : RedisBackend[T], _id : String) -> Unit

    ack confirms message processing.

    Current Valkey backend pop semantics remove the message from the list when BLPOP returns, so ack is a no-op until visibility-timeout tracking is added.

    RedisBackend::nack

    source
    async fn[T] RedisBackend::nack(_self : RedisBackend[T], _id : String) -> Unit

    nack signals processing failure.

    Current Valkey backend does not retain the popped payload by id, so nack is a no-op until in-flight tracking is added.

    RedisBackend::new

    source
    fn[T] RedisBackend::new(client : 
    Client    , key : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisBackend[T]

    new creates a RedisBackend instance with the provided client and serializers.

    RedisBackend::pop

    source
    async fn[T] RedisBackend::pop(self : RedisBackend[T]) -> Message[T]

    pop extracts a message from the Redis queue in a blocking manner using BLPOP.

    RedisBackend::pop_nowait

    source
    async fn[T] RedisBackend::pop_nowait(self : RedisBackend[T]) -> Message[T]?

    pop_nowait extracts a message without blocking, returning None when empty.

    RedisBackend::push

    source
    async fn[T] RedisBackend::push(self : RedisBackend[T], v : T) -> Unit

    push sends a message to the Redis queue using RPUSH.

    RedisBackend::size

    source
    async fn[T] RedisBackend::size(self : RedisBackend[T]) -> Int

    size returns the current number of queued messages.

    RedisBackend::to_relay_queue

    source
    fn[T] RedisBackend::to_relay_queue(self : RedisBackend[T]) -> RelayQueue[T]

    to_relay_queue exports the backend as a generic RelayQueue.

    RedisStreamBackend

    source
    pub struct RedisStreamBackend[T] {
      client : 
    Client
      stream : String
      dlq_stream : String
      group : String
      consumer : String
      serialize : (T) -> String
      deserialize : (String) -> T
    }

    RedisStreamBackend::ack

    source
    async fn[T] RedisStreamBackend::ack(self : RedisStreamBackend[T], id : String) -> Unit

    ack acknowledges a stream message by id.

    RedisStreamBackend::dead_letter

    source
    async fn[T] RedisStreamBackend::dead_letter(self : RedisStreamBackend[T], value : T) -> Unit

    dead_letter writes a payload to the configured DLQ stream.

    RedisStreamBackend::init

    source
    async fn[T] RedisStreamBackend::init(self : RedisStreamBackend[T]) -> Unit

    RedisStreamBackend::nack

    source
    async fn[T] RedisStreamBackend::nack(_self : RedisStreamBackend[T], _id : String) -> Unit

    nack currently leaves the message pending when only an id is available.

    Use nack_message when the payload is available and should be moved to DLQ.

    RedisStreamBackend::nack_message

    source
    async fn[T] RedisStreamBackend::nack_message(self : RedisStreamBackend[T], msg : Message[T]) -> Unit

    nack_message moves the message payload to the DLQ and acknowledges the original stream entry.

    RedisStreamBackend::new

    source
    fn[T] RedisStreamBackend::new(client : 
    Client    , stream : String, group : String, consumer : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisStreamBackend[T]

    RedisStreamBackend::new_with_dlq

    source
    fn[T] RedisStreamBackend::new_with_dlq(client : 
    Client    , stream : String, dlq_stream : String, group : String, consumer : String, serialize~ : (T) -> String, deserialize~ : (String) -> T) -> RedisStreamBackend[T]

    RedisStreamBackend::pending_count

    source
    async fn[T] RedisStreamBackend::pending_count(self : RedisStreamBackend[T]) -> Int64

    pending_count returns the number of unacknowledged messages.

    RedisStreamBackend::pop

    source
    async fn[T] RedisStreamBackend::pop(self : RedisStreamBackend[T]) -> Message[T]

    pop reads one message from the stream consumer group.

    RedisStreamBackend::pop_block

    source
    async fn[T] RedisStreamBackend::pop_block(self : RedisStreamBackend[T], block_ms : Int) -> Message[T]

    pop_block waits for a message for up to block_ms milliseconds.

    RedisStreamBackend::push

    source
    async fn[T] RedisStreamBackend::push(self : RedisStreamBackend[T], value : T) -> Unit

    RedisStreamBackend::recover_or_pop

    source
    async fn[T] RedisStreamBackend::recover_or_pop(self : RedisStreamBackend[T], min_idle_ms : Int, block_ms : Int) -> Message[T]

    recover_or_pop first tries to recover a pending message, then falls back to blocking reads for new stream entries.

    RedisStreamBackend::recover_pending

    source
    async fn[T] RedisStreamBackend::recover_pending(self : RedisStreamBackend[T], min_idle_ms : Int) -> Message[T]?

    recover_pending claims one pending message idle for at least min_idle_ms.

    RedisStreamBackend::to_relay_queue

    source
    fn[T] RedisStreamBackend::to_relay_queue(self : RedisStreamBackend[T]) -> RelayQueue[T]

    to_relay_queue exports the stream backend as a generic RelayQueue.

    RedisStreamBackend::to_relay_queue_with_recovery

    source
    fn[T] RedisStreamBackend::to_relay_queue_with_recovery(self : RedisStreamBackend[T], min_idle_ms : Int, block_ms : Int) -> RelayQueue[T]

    to_relay_queue_with_recovery exports the stream backend as a RelayQueue whose pop path first recovers pending entries before blocking for new ones.

    RelayQueue

    source
    pub struct RelayQueue[T] {
      push_fn : async (T) -> Unit
      pop_fn : async () -> Message[T]
      ack_fn : async (String) -> Unit
      nack_fn : async (String) -> Unit
      size_fn : () -> Int
    }

    RelayQueue is the main entry point for interacting with the message queue. It uses a closure-based backend abstraction to support multiple storage engines (Memory, Redis, etc.).

    RelayQueue::ack

    source
    async fn[T] RelayQueue::ack(self : RelayQueue[T], id : String) -> Unit

    ack confirms that a message has been processed successfully. The backend will permanently remove the message from its tracking system.

    RelayQueue::nack

    source
    async fn[T] RelayQueue::nack(self : RelayQueue[T], id : String) -> Unit

    nack (negative acknowledgement) informs the queue that processing failed. The backend will typically re-queue the message for another attempt, unless it exceeds the RetryPolicy.

    RelayQueue::new

    source
    fn[T] RelayQueue::new(push~ : async (T) -> Unit, pop~ : async () -> Message[T], ack~ : async (String) -> Unit, nack~ : async (String) -> Unit, size~ : () -> Int) -> RelayQueue[T]

    new creates a new RelayQueue with the provided backend functions.

    RelayQueue::pop

    source
    async fn[T] RelayQueue::pop(self : RelayQueue[T]) -> Message[T]

    pop removes and returns a value from the queue. Suspends if the queue is empty. In V3, a popped message is considered "In Flight" until acknowledged.

    RelayQueue::push

    source
    async fn[T] RelayQueue::push(self : RelayQueue[T], value : T) -> Unit

    push adds a value to the queue. Asynchronous and thread-safe via the underlying backend.

    RelayQueue::size

    source
    fn[T] RelayQueue::size(self : RelayQueue[T]) -> Int

    size returns the current number of items in the queue.

    RetryPolicy

    source
    pub struct RetryPolicy {
      max_attempts : Int
    }

    RetryPolicy defines how the queue handles failed message processing.

    RetryPolicy::default

    source
    fn RetryPolicy::default() -> RetryPolicy

    default_policy provides a standard policy with 3 max attempts.

    RetryPolicy::new

    source
    fn RetryPolicy::new(max_attempts : Int) -> RetryPolicy

    new_policy creates a custom retry policy.

    WorkerPool

    source
    pub struct WorkerPool[T] {
      queue : RelayQueue[T]
      concurrency : Int
    }

    WorkerPool manages a group of concurrent workers that process messages from the queue.

    WorkerPool::new

    source
    fn[T] WorkerPool::new(queue : RelayQueue[T], concurrency~ : Int) -> WorkerPool[T]

    new creates a new worker pool with the specified concurrency.

    WorkerPool::run

    source
    async fn[T] WorkerPool::run(self : WorkerPool[T], handler : async (T) -> Unit) -> Unit

    run starts the pool. Each worker is launched as a background task within a TaskGroup.

    start_worker

    source
    async fn[T] start_worker(queue : RelayQueue[T], handler : (Message[T]) -> Unit) -> Unit

    start_worker starts an infinite loop that pops messages from the queue and passes them to a handler.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io