Responders

A responder answers at the reply-to address the ROUTER stamps on each request, through a transform that names the destination.

A responder is an ordinary reply handler on a ZmqRpc subscription. Its ROUTER socket reads the identity of the peer behind each request and stamps the request with a reply-to header that addresses that peer, zmq-reply: and the identity in hex. The reply publisher routes an answer to such an address only. It refuses any other name with ZmqError::Send, so an answer that misses its address fails where it is sent.

A publish transform on the reply position sets the destination per request. It reads the delivery being answered, so it implements PublishTransform<ForReply<C>, Options>, and it names the destination, so it declares Destination = Names. It copies reply-to into the destination and correlation-id into the answer's headers, so the requester matches the answer to its request. get_shared hands each header over as the buffer the request arrived in, so answering costs a reference count rather than a copy. The answer type derives Outgoing without a name: a position offers the naming right only where the type declares no destination. The name in reply("reply") is the fallback that the document reports.

An answer whose requester has gone is refused with ZmqError::Send, naming the reply address: the ROUTER holds no peer under that identity. A responder addresses no retry copies of its own, so every responder mount names where they go with .out_retry(Publish).to("..").

Request and reply builds a scoring responder with this transform.

Example

The greeter answers each request at the peer that sent it.

Macros

examples/zmq_request_reply.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),
    }
}

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

// The transform reads the delivery being answered, so it names `ForReply`, and it names the
// destination per delivery, so it declares `Names`. Both headers are taken with `get_shared`,
// which hands them over as the buffer the request arrived in.
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);
        }
    }
}

Manual

examples/manual/zmq_request_reply.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),
        }))
    }
}

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

// The transform reads the delivery being answered, so it names `ForReply`, and it names the
// destination per delivery, so it declares `Names`. Both headers are taken with `get_shared`,
// which hands them over as the buffer the request arrived in.
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);
        }
    }
}

The mount puts the transform on the reply position.

Macros

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();
            },
        )
}

Manual

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();
            },
        )
}