Request-reply

request sends over a DEALER of its own and resolves with the answer matching its correlation-id, or RequestTimeout.

ZmqRpcPublisher implements the core's RequestReply capability, and it is the only publisher of the crate that does. A handler bounds its slot Out<impl RequestReply>, and the mount site binds ZmqRpcPublish to it. request(message, timeout) takes an OutgoingMessage of bytes the handler encoded, and its name is the name frame of the request. The request goes to the broker directly, past the slot's transforms and the app's publish layers.

Each request opens a DEALER socket of its own and dials the broker's endpoint, with the frames Envelope shows. The dial waits as long as the timeout allows, ten seconds at most, and a failed dial returns ZmqError::Endpoint. The request carries a correlation-id header: the one the caller set, or a fresh one. The DEALER keeps reading until an answer carries the same id, and the call returns that answer as a ZmqMessage, which the handler decodes. With no matching answer within the timeout, the call returns ZmqError::RequestTimeout. The responder on the other side copies the id onto its answer.

The timeout is the way to give up on a request: dropping the call closes its DEALER, as Cancellation describes. After shutdown a request returns ZmqError::NotConnected.

Request and reply asks a scoring service about each order from inside a handler.

Example

A handler on a queue asks the greeter for each name it receives, and acknowledges once the answer came back.

Macros
Manual

examples/zmq_request_reply.rs
#[subscriber("names")]
async fn ask(name: &Name, Out(greeter): Out<impl RequestReply>) -> HandlerOutcome {
    let Ok(greeting) = serde_json::to_vec(&Greeting {
        who: name.who.clone(),
    }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("greeter", greeting.as_slice());
    let Ok(answer) = greeter.request(request, Duration::from_secs(5)).await else {
        return HandlerOutcome::retry_after(Duration::from_secs(1));
    };
    let Ok(answer) = serde_json::from_slice::<Answer>(answer.payload()) else {
        return HandlerOutcome::drop();
    };
    tracing::info!(answer = %answer.text, "greeted");
    HandlerOutcome::ack()
}

examples/manual/zmq_request_reply.rs
struct Ask;

impl<Entry> Handle<Name, (), Outs<(Entry,)>> for Ask
where
    Entry: OutEntry<DefaultSlot, Wire: RequestReply>,
{
    async fn handle(
        &self,
        name: &Name,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let greeting = serde_json::to_vec(&Greeting {
            who: name.who.clone(),
        })
        .map_err(|_| HandlerOutcome::drop())?;
        let request = OutgoingMessage::new("greeter", greeting.as_slice());
        let answer = outs
            .get(DefaultSlot)
            .request(request, Duration::from_secs(5))
            .await
            .map_err(|_| HandlerOutcome::retry_after(Duration::from_secs(1)))?;
        let answer = serde_json::from_slice::<Answer>(answer.payload())
            .map_err(|_| HandlerOutcome::drop())?;
        tracing::info!(answer = %answer.text, "greeted");
        Ok(())
    }
}

The greeter here runs in the same service, on a ROUTER it binds, so the requests dial tcp://0.0.0.0:5557.

Macros
Manual

examples/zmq_request_reply.rs
#[ruststream::app]
fn app() -> impl App {
    let rpc = ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")).bindable();
    let to_greeter = rpc.bind(ZmqRpcPublish);
    RustStream::new(AppInfo::new("greeter", "0.1.0"))
        .with_broker(rpc, |b| {
            // A responder addresses no retry copies, so the mount names where they go.
            b.include(greet)
                .out_reply(ZmqRpcPublish)
                .transform(ReplyToRequester)
                .out_retry(ZmqRpcPublish)
                .to("greeter.retry");
        })
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(ask).out(DefaultSlot, to_greeter).build();
            },
        )
}

examples/manual/zmq_request_reply.rs
fn app() -> impl App {
    let rpc = ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")).bindable();
    let to_greeter = rpc.bind(ZmqRpcPublish);
    RustStream::new(AppInfo::new("greeter", "0.1.0"))
        .with_broker(rpc, |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(ZmqRpcPublish)
                .transform(ReplyToRequester)
                .out_retry(ZmqRpcPublish)
                .to("greeter.retry");
        })
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(subscriber("names", Ask).build())
                    .out(DefaultSlot, to_greeter)
                    .build();
            },
        )
}

Example​

Example