Subscriptions

A stream key in #[subscriber] is a subscription on both brokers; on a stream file it follows the live tail of that key.

#[subscriber("orders")] reads the stream key orders. A stream file holds many keys side by side, and a line on a pipe names its key in its header, so on both brokers the key is the whole subscription. On a stream file the subscription opens at the tip and reads every message appended under its key from then on. What the file held before the service started is read from a position or in a replay. A key is 1 to 249 characters of ASCII letters, digits, ., _ and -; a subscription to any other key stops the service at startup with an error that names the key. On a stream file, the descriptor FileStream opens the same subscription and adds the replay mode and a channel description in the generated document. On a pipe the key is the only form, and Lines shows how a line names it. The manual form passes the key to subscriber(name, body).

Example

Every order appended under orders while the service runs is shipped.

Macros

examples/file_subscriptions.rs

/// Ships every order appended under the `orders` key while the service runs.
#[subscriber("orders")]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "shipping");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            b.include(ship);
        },
    )
}

Manual

examples/manual/file_subscriptions.rs

/// Ships every order appended under the `orders` key while the service runs.
struct Ship;

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

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            b.include(subscriber("orders", Ship).build());
        },
    )
}