Reply address

A responder's reply channel has no address; the document points a client at $message.header#/reply-to instead.

A responder answers each request at the peer that sent it, so no fixed name addresses its answers. When the reply position carries a transform that names the destination, as a responder does, the document reports the reply channel with address: null. The receive operation's reply.address.location is $message.header#/reply-to, the header the ROUTER stamps on each request. The channel keeps the fallback name from reply("..") as its key.

Peers in other languages asks such a responder from a Python DEALER.

Example

The greeter's answer names no destination, and a transform names it per request. The mount on the AsyncAPI page puts the transform on the reply position.

Macros

examples/zmq_asyncapi.rs

// The ROUTER addresses each answer per request, so `reply` is only the fallback name.
#[derive(Debug, Serialize, Outgoing)]
struct Answer {
    text: String,
}

#[subscriber("greeter", reply("reply"))]
async fn greet(request: &Greeting) -> Answer {
    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;

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_asyncapi.rs

#[derive(Debug, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Answer {
    text: String,
}

// The ROUTER addresses each answer per request, so `reply` is only the fallback name.
impl OutgoingDestination for Answer {
    type Form = CallerName;
}

impl MessageHeaders for Answer {
    type Contract = NoHeaders;
}

struct Greet;

impl Handle<Greeting, Answer> for Greet {
    fn handle(
        &self,
        request: &Greeting,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Answer, HandlerOutcome>> {
        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;

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 reply channel and the operation that answers on it:

tests/documents/reply-address.json

{
  "channels": {
    "reply": {
      "address": null,
      "bindings": {
        "x-ruststream-zeromq": { "socketPair": "DEALER/ROUTER" }
      }
    }
  },
  "operations": {
    "receive_greeter": {
      "reply": {
        "address": { "location": "$message.header#/reply-to" }
      }
    }
  }
}