Publish

PulsarPublish, Publish in the prelude, opens one producer per topic and waits for the broker's receipt, so Ok means stored.

A mount site names it wherever a registration publishes: .out_reply(Publish) for a reply, .out(Slot, Publish) for a slot. The policy holds no connection; the runtime pairs it with the connected broker at startup into a PulsarPublisher. A registration that names no policy for its reply publishes through this one. A reply goes to the topic its type names with #[outgoing(name = "..")], or to the one a reply("..") clause names when the type leaves it open. The publisher opens a producer for a topic on the first publish there and keeps it until shutdown. A publish returns once the broker has stored the message, or with the error it reported. Outside a handler, PulsarBroker::publisher hands out a publisher before the service starts. Per message, a publish carries headers and a key.

Publishing and replies confirms orders with receipts and answers price quotes.

Example

Macros

examples/pulsar_publishing_publish.rs

/// The receipt of an order; the type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

/// Confirms each order with a receipt.
#[subscriber(PulsarSubscription::new("orders", "confirmations"), reply)]
async fn confirm(order: &Order) -> Receipt {
    tracing::info!(order.id, "order confirmed");
    Receipt { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(confirm).out_reply(Publish);
        },
    )
}

Manual

examples/manual/pulsar_publishing_publish.rs

/// The receipt of an order; the type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

/// Confirms each order with a receipt.
struct Confirm;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            let confirm = subscriber(PulsarSubscription::new("orders", "confirmations"), Confirm);
            b.include(confirm.reply().build()).out_reply(Publish);
        },
    )
}