ZmqRpc

ZmqRpc is DEALER/ROUTER: a responder answers each request at the peer that sent it, on whichever side the endpoint takes.

A responder is a subscription on a ROUTER socket. The ROUTER stamps each request with a reply-to header that addresses the peer that sent it. A publish transform sets the answer's destination to that address, so the reply type declares no destination of its own, and the name in reply("..") is the fallback the AsyncAPI document reports. The responders page shows the transform. An answer whose requester has gone is refused with ZmqError::Send, naming the reply address.

ZmqRpc carries no side in its type: ZmqRpc::new takes an endpoint that binds or one that dials. One policy, ZmqRpcPublish, serves both directions. It routes answers through the responder's ROUTER, and it sends requests over a DEALER socket of its own through the RequestReply capability. rpc::prelude offers it as Publish, together with RequestReply.

A responder answers each request at its own address, so .batch(..) on a responder does not compile.

A copy of a request has no address of its own, so a responder subscription addresses no retry copies. Every responder mount names where they go, with .out_retry(policy).to(".."), and a mount that names nothing fails startup with an error that names the subscription. A plain name reaches no peer on DEALER/ROUTER. A service that wants the copies to arrive sends them through a queue on another broker; a service that does not names a destination to say so and lets the requester ask again.

Request and reply builds a scoring responder and a checkout service that asks it.

Example

A greeter binds tcp://0.0.0.0:5557 and answers each request at the peer that asked. Its retry copies go to greeter.retry, a name that reaches no peer: a requester that gets no answer asks again.

Macros
Manual

examples/zmq_broker_rpc.rs
// The answer goes to the peer that asked, so the clause's name is only the fallback.
#[subscriber("greeter", reply("reply"))]
async fn greet(request: &Greeting) -> Answer {
    tracing::info!(who = %request.who, "greeting");
    Answer {
        text: format!("hello {}", request.who),
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("greeter", "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(greet)
                .out_reply(Publish)
                .transform(ReplyToRequester)
                .out_retry(Publish)
                .to("greeter.retry");
        },
    )
}

examples/manual/zmq_broker_rpc.rs
struct Greet;

impl Handle<Greeting, Answer> for Greet {
    fn handle(
        &self,
        request: &Greeting,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Answer, HandlerOutcome>> {
        tracing::info!(who = %request.who, "greeting");
        ready(Ok(Answer {
            text: format!("hello {}", request.who),
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("greeter", "0.1.0")).with_broker(
        ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            // The answer goes to the peer that asked, so the `to` name is only the fallback.
            // A responder addresses no retry copies, so the mount names where they go.
            b.include(subscriber("greeter", Greet).reply().to("reply").build())
                .out_reply(Publish)
                .transform(ReplyToRequester)
                .out_retry(Publish)
                .to("greeter.retry");
        },
    )
}

Example​

Example