Memory broker

MemoryBroker is a complete broker inside the service's process, and the reference implementation of the broker contract.

It comes with the memory feature and needs no server. It suits a queue that belongs to one process, a prototype, or local development; the default cargo generate template starts on it. MemoryBroker::new does no I/O, like every broker constructor. A broker built with it keeps nothing: a message lives until the last subscriber has read it, so memory holds the work the handlers have yet to do. A broker that keeps the newest messages for replay is built with retaining. A mount site imports ruststream::memory::prelude, one glob with the core prelude, the broker's own types, and its publish policies under the names every broker crate uses: Publish, TransactionalPublish and Request. Partitioned stays out of the glob, because in scope it makes msg.partition_key() ambiguous; a service that reads partition keys imports it. A handler body that names capabilities, not policies, keeps ruststream::prelude::*.

A subscription by name reads messages published to that exact name, from the moment it opens. Every subscription of a name receives every message published to it. MemorySource is the broker's subscription descriptor, and MemoryPattern reads every name a pattern matches. An ack or a drop consumes a delivery; a retry, or a delivery left without a settlement, comes back to the same subscriber. A retry_after delivery comes back to the same subscriber after its delay, and nothing is republished. Every delivery reports how many times the broker has handed it to that subscriber, so a registration's attempt cap counts these redeliveries. MemoryBroker is a shared handle: its clones work on one bus. After shutdown, a publish, a commit or a request through a handle taken earlier returns an error.

Example

The dependency line:

ruststream = { version = "0.7", features = ["macros", "memory", "json"] }

Two subscriptions to orders each receive the order, and one of them replies over the same bus.

Macros

examples/memory_broker.rs

use ruststream::memory::prelude::*;

#[subscriber("orders", reply("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[subscriber("orders")]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(confirm);
        b.include(audit);
    })
}

Manual

examples/manual/memory_broker.rs

use ruststream::memory::prelude::*;

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation { id: order.id }))
    }
}

struct Audit;

impl Handle<Order> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "audited");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .to("confirmations")
                .build(),
        );
        b.include(subscriber("orders", Audit).build());
    })
}

Capabilities

The broker implements the optional capabilities on its own in-process semantics.

• Request and reply: broker.requester() gives a MemoryRequester, which names a unique reply name in the reply-to header and waits for the first message delivered there. Request is its policy.
• Batches: a batch is the first delivery to arrive plus everything already buffered, up to the size the registration names.
• Transactions: the publisher behind Publish implements TransactionalPublisher and OwnedTransactions. A commit delivers the buffered messages at once, in publish order; an abort discards them.
• Partition keys: a MemoryMessage reads its key from the partition-key header.
• Replay: on a retaining broker, a subscription opens at a start position and a handler seeks.

The TestApp harness runs an application on this broker as it runs one on a networked broker, and its publishes wait on exactly the subscriptions the broker's routing picks.