PubSubPublish

PubSubPublish, Publish in the prelude, is the publish policy; every publish waits for the message id Pub/Sub assigns.

A mount site names it wherever a registration publishes: .out_reply(Publish::default()) for a reply, .out(Slot, Publish::default()) for a slot. The policy holds no connection; the runtime pairs it with the connected broker at startup into a PubSubPublisher. A registration that names no policy for its reply publishes through this one. The destination is a topic, a short id in the broker's project or a full resource name. A reply goes to the topic its type names, and a slot publish to the topic the call or the slot names.

The encoded message becomes the Pub/Sub message's data without a copy, and headers become its attributes. The client library keeps one publisher per topic, created on first use, which sends messages in batches of up to 100 messages, 1 MB or 10 milliseconds. A publish returns once the service has accepted the message and assigned its id, or with the error it reported as PubSubError::Publish. The policy's one setting is a default ordering key.

Example

A reply goes to the topic of its type, and an invoice to the topic the call names.

Macros
Manual

examples/pubsub_publishing.rs
/// The receipt goes to `receipts`, the topic its type names.
#[subscriber(GooglePubSub::new("orders-receipts").create_with_topic("orders"), reply)]
async fn receipt(order: &Order) -> Receipt {
    Receipt { order_id: order.id }
}

examples/manual/pubsub_publishing.rs
/// The receipt goes to `receipts`, the topic its type names.
struct ReceiptFor;

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

Macros
Manual

examples/pubsub_publishing.rs
/// The invoice goes to the topic the call names.
#[subscriber(GooglePubSub::new("orders-billing").create_with_topic("orders"))]
async fn bill(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let invoice = Invoice { order_id: order.id };
    match out.message(&invoice).to("invoices").publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, error = %err, "invoice not published");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/pubsub_publishing.rs
/// The invoice goes to the topic the call names.
struct Bill;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Bill
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let invoice = Invoice { order_id: order.id };
        let sent = outs
            .get(DefaultSlot)
            .message(&invoice)
            .to("invoices")
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(order.id, error = %err, "invoice not published");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

The mount site names the policy for both:

Macros
Manual

examples/pubsub_publishing.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(receipt).out_reply(Publish::default());
            b.include(bill).out(DefaultSlot, Publish::default()).build();
        },
    )
}

examples/manual/pubsub_publishing.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-receipts").create_with_topic("orders"),
                    ReceiptFor,
                )
                .reply()
                .build(),
            )
            .out_reply(Publish::default());
            b.include(
                subscriber(
                    GooglePubSub::new("orders-billing").create_with_topic("orders"),
                    Bill,
                )
                .build(),
            )
            .out(DefaultSlot, Publish::default())
            .build();
        },
    )
}

Example​

Example