Skip to main content

Publishing and replies

This guide replies to Pulsar messages, publishes keyed messages with headers through an Out slot, and publishes once the service has started.

The example is a checkout service. It confirms orders with receipts, answers price quotes, sends pick lists to warehouses and announces itself at startup.

A reply to a declared topic​

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

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

confirm returns a Receipt, and the framework publishes it. #[outgoing(name = "receipts")] fixes the topic on the type, so the subscriber writes the bare reply clause; in the manual form, .reply(). The broker's publish policy, Publish in the prelude, is its default, so a plain include replies through it. A publish returns once the broker has stored the message.

A reply to a topic the mount site names​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct QuoteRequest {
    request: u64,
    items: u32,
}

/// A price quote; the mount site names where it goes.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Quote {
    request: u64,
    total_cents: u64,
}

/// Answers a quote request on `quotes`, the topic the requesting services read.
#[subscriber(PulsarSubscription::new("quote-requests", "quotes"), reply("quotes"))]
async fn quote(request: &QuoteRequest) -> Quote {
    Quote {
        request: request.request,
        total_cents: u64::from(request.items) * 1_250,
    }
}

Quote names no topic, so reply("quotes") names it at the mount site; in the manual form, .reply().to("quotes"). The same type can then answer on another topic in another service.

A request and its answer travel on two ordinary topics. A service that asks for quotes publishes to quote-requests and reads quotes under a subscription of its own. It matches each quote to its request by the request id the two messages share.

A keyed publish with headers​

src/main.rs
/// A pick list for the warehouse that ships the order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "warehouse-picks")]
struct Pick {
    order: u64,
}

#[derive(OutSlot)]
#[publishes(Pick)]
struct Picks;

/// Sends each order to its warehouse, keyed by the warehouse so its picks stay in order.
#[subscriber(PulsarSubscription::new("orders", "dispatch"))]
async fn dispatch(
    order: &Order,
    Out(picks): Out<impl Publisher<Options = PulsarPublishOptions>, Picks>,
) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("source", "checkout");
    let sent = picks
        .message(&Pick { order: order.id })
        .with_headers(headers)
        .partition_key(order.warehouse.clone())
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(order.id, %error, "pick list not sent");
            HandlerOutcome::retry()
        }
    }
}

An Out slot publishes from the handler's body, and Picks declares the type it carries. partition_key keys each pick list by its warehouse. On a partitioned topic the key picks the partition, and a KeyShared subscription keeps the order of each key. The step comes from the broker's prelude, so the slot is bounded Publisher<Options = PulsarPublishOptions> to offer it.

Each header becomes a message property of the same name. A consumer in another language reads source as an ordinary Pulsar property, with no envelope around the payload. Pulsar carries properties as text, so a header value that is not UTF-8 fails the publish.

A publish that fails returns the error, and dispatch retries the order. A payload larger than the broker's max_message_size, five mebibytes by default, fails before it is sent.

The app​

src/main.rs
/// Tells the rest of the system that checkout is up.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "service-events")]
struct Started {
    service: String,
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(confirm);
            b.include(quote);
            b.include(dispatch).out(Picks, Publish).build();
            b.after_startup(Publish, async move |publisher| {
                let started = Started {
                    service: "checkout".to_owned(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}

.out(Picks, Publish) wires the slot to the publish policy. after_startup runs once the subscriptions are open, with a publisher of the policy it names, and checkout announces itself on service-events. A hook that fails stops the service at startup.

The publisher opens one producer per topic on the first publish to it, and shutdown closes them. Code outside the app takes a publisher from the broker: PulsarBroker::publisher before the service starts, and ConnectedPulsarBroker::publisher after.

Testing the publishes​

The tests in the example file run the app in process. published checks what reached a topic, with its headers, and tb.out::<Picks>() checks the key the slot's publish carried. The in-process Pulsar frames headers and the key the way the client does, so a header the server would refuse fails the test too.