Memory

MemoryBroker, behind the memory feature, is a complete broker that runs inside your process. It suits a queue that belongs to a single application rather than to a network. The default cargo generate template (templates/memory) uses it, so a fresh project runs with no external dependencies.

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

use ruststream::memory::MemoryBroker;

let broker = MemoryBroker::new();

How much it keeps

A broker built with MemoryBroker::new() keeps nothing. A message lives from the publish until the last subscriber has read it, so the memory a long-running service holds is the work its handlers have yet to do, whatever the message count.

A service that replays needs history, and says how much of it to keep:

examples/seek.rs

// Replaying reads what the broker kept, so this one keeps the last 64 entries of every log.
// `MemoryBroker::new()` keeps nothing, and a mount that seeks does not compile on it.
let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(64)));

Retention bounds one topic: Messages(n) keeps the newest n messages of every topic, Bytes(n) keeps the newest payloads that fit in n bytes, and MessagesAndBytes { .. } applies both at once. A broker publishing under a thousand topics therefore holds up to that much for each of them. The newest message always stays, so a payload wider than a byte bound is kept alone rather than dropped when it arrives.

The two forms are different types: MemoryBroker::retaining(..) gives the one whose subscriptions are Seekable, and a mount that opens at a position or reads a seek handle only compiles against it. Replaying on a broker that keeps nothing is a compile error, not a replay that quietly finds nothing.

The prelude a mount site imports

ruststream::memory::prelude is this broker's glob, built like the prelude of every broker crate. It re-exports the core prelude, then the broker's own surface (MemoryBroker, MemorySource, MemoryError, MemoryPosition, Retention with the log modes Discarding / Retaining, and the context keys MemoryContext / MemoryBatchContext / Position / SeekHandle), then the publish policies under the uniform names Publish, TransactionalPublish and Request. All three are aliases of MemoryPublish and MemoryRequest. This broker's publisher implements both transaction kinds, so TransactionalPublish here is the same policy as Publish; on a broker with a separate transactional configuration that name points to a different policy.

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

The same glob brings in the capability traits this broker implements (TransactionalPublisher, OwnedTransactions, Transaction, RequestReply, Positioned, Seeker), so their operations are in scope wherever the policies are. Partitioned stays out: in scope it makes msg.partition_key() ambiguous with the method of the same name on IncomingMessage. A service that reads partition keys imports Partitioned itself.

A handler body keeps use ruststream::prelude::*;: it names capabilities, never policies, and does not know which broker runs it. A file holding both a body and its mount site needs the broker glob alone.

Semantics

• Topic names match in full. A subscription to orders receives the messages published to orders.
• Fan-out. Every subscriber of a topic receives every message published to it after the subscription.
• Ack is a no-op; nack(requeue: true) redelivers the same payload to the same subscriber.
• retry_after is the broker's own. The delivery comes back to the same subscriber once the delay has elapsed, and nothing is republished in the meantime.
• Deliveries are counted. Every delivery reports how many times the broker has handed that subscriber the message, the first one included, so a registration's max_attempts(n) is spent on these redeliveries and the delivery that spends it reaches the dead_letter(name) destination.
• Shared ownership. MemoryBroker is a reference-counted handle: all its owners work with one state, so a clone held by a test sees everything the application publishes.

A handler, its middleware and its decoding behave here as they do against a networked broker: the runtime dispatches messages through the same path.

Capabilities

Every capability trait is implemented over this broker's own in-process semantics:

• Request / reply. broker.requester() gives you a MemoryRequester: its request publishes the message and names a unique in-process reply topic in the reply-to header, and completes with the first message delivered there. The responder reads reply-to from the request and publishes its reply to that topic. A request nobody answers returns RequestError::Timeout. MemoryRequest is the policy that constructs MemoryRequester, so you bind a slot bound with Out<impl RequestReply, ..> to MemoryRequest.
• Batches. MemorySubscriber implements BatchSubscriber: a batch is the first delivery to arrive plus everything already buffered, capped at the size the handler registration named with batch(n). A partial batch is delivered immediately.
• Transactions. MemoryPublish is the policy that constructs MemoryPublisher, which implements both transaction kinds, so you bind a slot or a wiring bound with TransactionalPublisher or OwnedTransactions to MemoryPublish. Publishes inside a transaction scope are buffered: commit delivers them to every subscriber at once in publish order, abort discards them. Every owned transaction buffers on its own, and clones of a publisher handle do not share its transaction. Out-of-order calls on the publisher itself return MemoryError: a second begin_transaction while one is open returns TransactionBusy and leaves the open transaction untouched, and a commit or abort without one returns NoTransaction.
• Partition keys. MemoryMessage implements Partitioned and reads the key from the partition-key header (memory::PARTITION_KEY_HEADER).
• Seeking. On a retaining broker, MemorySubscriber implements Seekable over the per-topic log: get a MemorySeeker before reading starts, then call seek with a MemoryPosition, taken from a delivered message with Positioned::position (which delivers that same message again) or constructed (MemoryPosition::start() / sequence(n) / end()). Sequence numbers are absolute and keep naming the same message as the retention bound evicts older ones. start() is the oldest message still kept and end() is the tip, past everything published so far. Seeking forward skips the deliveries queued before the target. A sequence the bound has already evicted returns MemoryError::PositionEvicted, which reports the oldest position left. A seek acts on one subscriber instance. Through a handle to a bus that has already shut down it returns MemoryError::ShutDown. Inside an application, MemoryContext holds the position of the message and the MemorySeeker, and a handler reads them under the Position and SeekHandle keys (see Seeking). A batch handler reads MemoryBatchContext: it holds SeekHandle but no Position, because a batch spans many deliveries.
• Shutdown. MemoryBroker::connect(self) gives ConnectedMemoryBroker, and its shutdown consumes self and returns ClosedMemoryBroker, which reports how many subscriber registrations the shutdown dropped. After that, a publish, a transaction commit or a request through a handle handed out earlier returns MemoryError::ShutDown or RequestError::ShutDown.

Subscription source

ConnectedMemoryBroker implements Subscribe, so #[subscriber("orders")] works directly. The MemorySource descriptor names the same subscription, in the form every broker uses. From the routed_service example:

MacrosManual

examples/routed_service/orders.rs

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

#[subscriber(MemorySource::new("orders"), publish("confirmations"))]
pub(crate) async fn confirm(
    order: &Order,
    ctx: &mut Context<'_, (), Repository>,
) -> Result<Confirmation, HandlerOutcome> {
    let repo = ctx.state();
    tracing::debug!(
        order = order.id,
        customer = %order.customer,
        item = %order.item,
        "confirming order"
    );
    match repo.record_order(order.id).await {
        Ok(()) => Ok(Confirmation {
            order_id: order.id,
            accepted: order.quantity > 0,
        }),
        Err(e) if e.is_transient() => {
            tracing::warn!(order = order.id, "store busy, asking for redelivery");
            Err(HandlerOutcome::retry())
        }
        Err(e) => {
            tracing::error!(order = order.id, error = %e, "dropping order");
            Err(HandlerOutcome::drop())
        }
    }
}

examples/manual/routed_service_orders.rs

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

struct Confirm;

// The state is named on the body, not on a definition: this one reads a `Repository`, so it is a
// `Handle` for that state alone and mounts only on an application that carries it.
impl Handle<Order, Confirmation, (), (), Repository> for Confirm {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), Repository>,
    ) -> Result<Confirmation, HandlerOutcome> {
        let repo = ctx.state();
        tracing::debug!(
            order = order.id,
            customer = %order.customer,
            item = %order.item,
            "confirming order"
        );
        match repo.record_order(order.id).await {
            Ok(()) => Ok(Confirmation {
                order_id: order.id,
                accepted: order.quantity > 0,
            }),
            Err(e) if e.is_transient() => {
                tracing::warn!(order = order.id, "store busy, asking for redelivery");
                Err(HandlerOutcome::retry())
            }
            Err(e) => {
                tracing::error!(order = order.id, error = %e, "dropping order");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

/// The mount, and the whole declaration the attribute's clauses carried: the broker's own
/// descriptor as the source, `.to(..)` for the reply channel, and `.describe(..)` for the sentence
/// the attribute lifts off the handler's doc comment. Who publishes the reply is wiring rather
/// than declaration, so it lives on the mount chain on both paths: `.out_reply(Publish)` names
/// the position the returned value leaves through and the policy that carries it, which pairs with
/// the connected broker at startup and encodes with the default codec. The definition says what it
/// replies with and where; the chain says who sends it.
fn confirm_route() -> impl RouterDef<MemoryBroker, Repository> {
    Router::<MemoryBroker>::new()
        .include(
            subscriber(MemorySource::new("orders"), Confirm)
                .reply()
                .to("confirmations")
                .describe("Confirms an order and replies on `confirmations`.")
                .build(),
        )
        .out_reply(Publish)
        .build()
}

For testing

You test an application built on MemoryBroker with the TestApp harness: build the app, hand it to TestApp::start, publish messages, and assert on what the handlers received and published. Testing walks through the full pattern.

The harness records what the service publishes for the length of a run, so published::<T>(..) assertions read the same list whichever form of the broker the application was built on. Outside the harness, reading a broker's log back through TestableBroker::published shows what that broker keeps: everything on a retaining one within its bound, nothing on the default one.