跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

请求与回复

本指南在 DEALER/ROUTER 上应答请求，并在处理器内部向另一个服务发出请求。

示例包含两个服务。 评分服务回答某个客户的订单金额是否获批。 结账服务为它接收的每个订单向评分服务发出询问。

应答方​

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 是 DEALER/ROUTER 模式。 应答方是稳定的一端，所以它绑定，每个请求方都拨号连接它。 score 是一个普通的回复处理器：它返回答案，或者丢弃没有金额的请求。

ROUTER 套接字知道每个请求来自哪个对端。 它把这个对端以 reply-to 消息头的形式标在请求上，答案必须发往那里。 因此 Score 不声明自己的目的地，reply("reply") 只指定一个后备目的地。 下面的转换为每个请求替换它。

应答方要指定它的重试副本发往何处：.out_retry(Publish).to("score.retry")。 请求的副本没有自己的地址，因此缺少这一步的挂载会在启动时被拒绝。 应答方也不接收批：在它上面调用 .batch(..) 无法编译，因为每个答案都发往各自的对端。

转换​

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 是作用于回复的发布转换。 它读取请求的 reply-to 消息头，把答案发往那里。 它还复制请求的 correlation-id，请求方依据它匹配答案。 它声明 Destination = Names，因为目的地由它选择。 只有在回复类型没有声明目的地的地方，转换才可以这样做。 因此带名称的回复类型和这个转换无法一起编译。

发给已经离开的对端的答案会被拒绝，错误中写明它的地址。 应答方由此得知它的答案无处可达；而在广播上，不匹配的消息会被悄无声息地丢弃。

请求方​

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 接收一个 Out<impl RequestReply> 参数，挂载点用 .out(DefaultSlot, ask) 把它绑定到一个 ZmqRpc 策略。 scoring Broker 拨号连接评分服务，没有订阅；它只发出请求。 request 通过自己的 DEALER 套接字发送请求，并返回带有相同 correlation-id 的答案。 check 自己编码请求、解码答案，所用的 JSON 与评分服务读写的一致。

超时​

在 ASK_TIMEOUT 内没有得到答案的请求以 ZmqError::RequestTimeout 失败。 这时 check 返回 retry_after(ASK_AGAIN)，订单在十秒后回来。 max_attempts 把尝试次数限制为三次。 这就是 ZeroMQ 指南中的 Lazy Pirate 模式，由框架负责计数。 超时的请求可能已经到达了应答方，因此请求必须可以安全地发送两次。

request 和回复发布都不是取消安全的。 处理器通过请求的超时来放弃，而不是通过 select! 分支（取消）。

测试​

评分服务的测试像请求方的 DEALER 那样发送请求，并检查 score 如何结算它们。 结账服务的测试在进程内运行，没有评分服务，因此它的请求会超时。 测试检查 check 要求了重试，并且副本在 ASK_AGAIN 之后回来，重试次数为一。 时钟是暂停的，因此超时和延迟瞬间就过去了。 测试 ZeroMQ 服务讲解了测试套件。