Skip to main content

PublishDynStack

A PublishDynStack is a publish middleware chain assembled when the app starts, added to the app as one publish layer.

A PublishDynStack holds a list chosen when the app is built: from configuration, a loop or a feature flag. Each element implements PublishDynLayer. It gets the outgoing message and a PublishDynNext; next.run(out) runs the rest of the list and then the static chain around the stack. PublishDynStack::new(list) freezes the list, and the first element runs outermost. The stack is a PublishLayer itself, added with RustStream::publish_layer, so it wraps the same publishes as any other. Each element costs one boxed future and one dynamic call per publish, and the stack two boxes more: its continuation and the future of the chain after it. The static layers around it stay free. The layer has no macro form: the same impl serves both forms, and only the handler differs.

Example​

The layer writes one header on every message.

examples/publish_dyn_stack.rs
/// Writes one header on every message the app publishes.
struct Stamp {
    // Both clone by reference count, so a publish copies no bytes.
    name: Str,
    value: Bytes,
}

impl PublishDynLayer for Stamp {
    fn on_publish<'a>(
        &'a self,
        out: &'a mut Outgoing<'a>,
        next: PublishDynNext<'a>,
    ) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'a>> {
        out.headers_mut()
            .insert(self.name.clone(), self.value.clone());
        next.run(out)
    }
}

The app builds one per header the configuration lists.

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

/// The chain comes from configuration: one `Stamp` per header the operator listed.
fn app(headers: &[(&str, &str)]) -> impl App {
    let chain = headers.iter().map(|(name, value)| {
        let stamp = Stamp {
            name: Str::from((*name).to_owned()),
            value: Bytes::from((*value).to_owned()),
        };
        Arc::new(stamp) as Arc<dyn PublishDynLayer>
    });
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        .publish_layer(PublishDynStack::new(chain))
        .with_broker(MemoryBroker::new(), |b| {
            b.include(issue_receipt);
        })
}