Retention

A retaining MemoryBroker keeps the newest messages of every name within a bound, and its subscriptions replay over them.

MemoryBroker::retaining takes a Retention bound, applied to each name on its own. Retention::Messages(n) keeps the newest n messages. Retention::Bytes(n) keeps the newest payloads that fit in n bytes. Retention::MessagesAndBytes { .. } applies both. A broker that publishes under a thousand names keeps up to that much for each of them. The newest message always stays, so a payload larger than a byte bound is kept alone. The retaining broker is a type of its own, MemoryBroker<Retaining>, and only its subscriptions implement Seekable. A start_at or a seek therefore does not compile on MemoryBroker::new(), which keeps nothing. A MemoryPosition names a place in the log of one name: start() is the oldest message still kept, sequence(n) the n-th message ever published to the name, and end() the tip. A sequence number keeps naming the same message as the bound evicts older ones. A sequence the bound has already evicted is refused with MemoryError::PositionEvicted, which names the oldest position left. A pattern subscription does not replay, because the log is kept per name.

Example

The broker keeps two messages of each name; three published before the service starts leave the last two to replay.

Macros
Manual

examples/memory_retention.rs
/// Keeps the newest two messages of every name.
fn broker() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(2)))
}

/// Opens at the oldest message still kept, so it reads what came before the service.
#[subscriber("audit", start_at(MemoryPosition::start()))]
async fn record(entry: &Entry) -> HandlerOutcome {
    tracing::info!(entry.id, "recorded");
    HandlerOutcome::ack()
}

fn app(broker: MemoryBroker<Retaining>) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        b.include(record);
    })
}

examples/manual/memory_retention.rs
/// Keeps the newest two messages of every name.
fn broker() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(2)))
}

struct Record;

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

fn app(broker: MemoryBroker<Retaining>) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // Opens at the oldest message still kept, so it reads what came before the service.
        b.include(
            subscriber("audit", Record)
                .start_at(MemoryPosition::start())
                .build(),
        );
    })
}

Example​

Example