Skip to main content

Request and reply

This guide answers requests on DEALER/ROUTER and asks another service from inside a handler.

The example has two services. The scoring service answers whether an order amount is approved for a customer. The checkout service asks it about each order it takes.

The responder​

src/main.rs
use ruststream_zeromq::rpc::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize)]
struct ScoreRequest {
    customer: u64,
    amount_cents: u64,
}

// No destination of its own: the ROUTER addresses each answer per request.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
struct Score {
    customer: u64,
    approved: bool,
}

// The answer goes to the peer that asked, so the clause's name is only the fallback.
#[subscriber("score", reply("reply"))]
async fn score(request: &ScoreRequest) -> Result<Score, HandlerOutcome> {
    if request.amount_cents == 0 {
        tracing::warn!(request.customer, "a request without an amount, dropped");
        return Err(HandlerOutcome::drop());
    }
    tracing::info!(request.customer, request.amount_cents, "scoring");
    Ok(Score {
        customer: request.customer,
        approved: request.amount_cents <= 50_000,
    })
}

#[ruststream::app]
fn app() -> impl App {
    // The responder is the stable side: it binds, and every requester dials it.
    RustStream::new(AppInfo::new("scoring", "0.1.0")).with_broker(
        ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            // A responder addresses no retry copies, so the mount names where they go.
            b.include(score)
                .out_reply(Publish)
                .transform(ReplyToRequester)
                .out_retry(Publish)
                .to("score.retry");
        },
    )
}

ZmqRpc is the DEALER/ROUTER pattern. The responder is the stable side, so it binds, and every requester dials it. score is an ordinary reply handler: it returns the answer, or drops a request without an amount.

A ROUTER socket knows which peer sent each request. It stamps that peer on the request as a reply-to header, and the answer must go there. So Score declares no destination of its own, and reply("reply") names only a fallback. The transform below replaces it for each request.

A responder names where its retry copies go: .out_retry(Publish).to("score.retry"). A copy of a request has no address of its own, so a mount without it is refused at startup. A responder also takes no batches: .batch(..) on it does not compile, because each answer goes to its own peer.

The transform​

src/main.rs
/// Answers at the `reply-to` address the ROUTER stamped on the request, under the request's
/// `correlation-id`.
struct ReplyToRequester;

impl<C, Options> PublishTransform<ForReply<C>, Options> for ReplyToRequester {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(reply_to) = cx.headers().get_shared("reply-to")
            && let Ok(reply_to) = Str::try_from(reply_to)
        {
            out.set_name(reply_to);
        }
        if let Some(correlation) = cx.headers().get_shared("correlation-id") {
            out.headers_mut()
                .insert(Str::from_static("correlation-id"), correlation);
        }
    }
}

ReplyToRequester is a publish transform on the reply. It reads the request's reply-to header and sends the answer there. It also copies the request's correlation-id, which the requester matches the answer by. It declares Destination = Names, because it chooses the destination. A transform may do that only where the reply type declares none, so a named reply type and this transform do not compile together.

An answer to a peer that has gone is refused with an error naming its address. The responder learns that its answer arrived nowhere; on a fan-out, an unmatched message is dropped without a word.

The requester​

src/main.rs
use std::time::Duration;

use ruststream::OutgoingMessage;
use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

/// How long one request waits for its answer.
const ASK_TIMEOUT: Duration = Duration::from_secs(2);
/// When the order is asked about again after the scoring service did not answer.
const ASK_AGAIN: Duration = Duration::from_secs(10);

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    customer: u64,
    amount_cents: u64,
}

#[derive(Serialize)]
struct ScoreRequest {
    customer: u64,
    amount_cents: u64,
}

#[derive(Deserialize)]
struct Score {
    approved: bool,
}

#[subscriber("orders")]
async fn check(order: &Order, Out(scoring): Out<impl RequestReply>) -> HandlerOutcome {
    let Ok(request) = serde_json::to_vec(&ScoreRequest {
        customer: order.customer,
        amount_cents: order.amount_cents,
    }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("score", request.as_slice());
    let answer = match scoring.request(request, ASK_TIMEOUT).await {
        Ok(answer) => answer,
        Err(err) => {
            tracing::warn!(order.id, %err, "no score, asking again later");
            return HandlerOutcome::retry_after(ASK_AGAIN);
        }
    };
    let Ok(score) = serde_json::from_slice::<Score>(answer.payload()) else {
        return HandlerOutcome::drop();
    };
    tracing::info!(order.id, score.approved, "order checked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The scoring service binds; every requester dials it.
    let scoring = ZmqRpc::new(ZmqEndpoint::connect("tcp://scoring:5557")).bindable();
    let ask = scoring.bind(ZmqRpcPublish);
    RustStream::new(AppInfo::new("checkout", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(check)
                    .max_attempts(nonzero!(3u32))
                    .out(DefaultSlot, ask)
                    .build();
            },
        )
        .with_broker(scoring, |_b| {})
}

check takes an Out<impl RequestReply> parameter, and the mount site binds it to a ZmqRpc policy with .out(DefaultSlot, ask). The scoring broker dials the scoring service and has no subscription; it only asks. request sends the request over a DEALER socket of its own and returns the answer that carries the same correlation-id. check encodes the request and decodes the answer itself, with the JSON the scoring service reads and writes.

Timeouts​

A request that gets no answer in ASK_TIMEOUT fails with ZmqError::RequestTimeout. check then returns retry_after(ASK_AGAIN), and the order comes back ten seconds later. max_attempts caps the tries at three. This is the Lazy Pirate pattern of the ZeroMQ guide, with the framework keeping the count. A request that timed out may still have reached the responder, so a request must be safe to send twice.

Neither request nor a reply publish is safe to cancel. A handler gives up through the request's timeout, not through a select! arm (Cancellation).

Testing​

The scoring service's tests send requests as a requester's DEALER would and check how score settled them. The checkout test runs with no scoring service in process, so its request times out. It checks that check asked for a retry, and that the copy came back after ASK_AGAIN with a retry count of one. The clock is paused, so the timeout and the delay pass at once. Testing a ZeroMQ service covers the harness.