PublishLayer

A publish layer wraps every message the app's handlers publish: it changes the message, observes the send, or refuses it.

A PublishLayer gets the Outgoing message and a PublishNext. next.run(out) runs the layers after it and then the broker send. Before that call the layer may change the message, its headers for one; after it, the layer sees the send's result. An error returned without the call refuses the publish. RustStream::publish_layer adds a layer to the app, before the first with_broker; the layer is Clone. It wraps a reply and every message sent through an Out slot. A message passes its position's transforms first, then the app's publish layers, then the send. The last publish layer added runs outermost, the reverse of the order of handler layers. A handler mounted through a router is the exception on the slot side: its slots take their own transforms, not the app's publish layers; its replies pass through them. The chain is one type, with no dynamic call. A layer has no macro form: the same impl serves both forms, and only the handler differs. A chain chosen when the app starts goes into a PublishDynStack.

Example

The layer names the service on every message it publishes.

examples/publish_layer.rs
/// Names the service on every message it publishes.
#[derive(Clone)]
struct Origin {
    service: &'static str,
}

impl PublishLayer for Origin {
    async fn on_publish<'a, N: PublishPipeline, P: Publisher>(
        &'a self,
        out: &'a mut Outgoing<'a>,
        next: PublishNext<'a, N, P>,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        out.headers_mut().insert("origin", self.service);
        next.run(out).await
    }
}

Macros
Manual

examples/publish_layer.rs
#[subscriber("orders", reply)]
async fn issue_receipt(order: &Order) -> Receipt {
    Receipt { order_id: order.id }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        // The replies and the `Out` slot publishes of the handlers below pass through it.
        .publish_layer(Origin { service: "billing" })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(issue_receipt);
        })
}

examples/manual/publish_layer.rs
struct IssueReceipt;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        // The replies and the `Out` slot publishes of the handlers below pass through it.
        .publish_layer(Origin { service: "billing" })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("orders", IssueReceipt).reply().build());
        })
}

Example​

Example