Replies

A queue worker replies through a queue on an endpoint of its own, since the one its subscription binds would hand the reply back to it.

A reply type that owns its destination names it with #[outgoing(name = "..")], and the subscriber clause stays a bare reply. A type that names nothing takes the name the clause gives, #[subscriber("jobs", reply("results"))]. The generated document reports the destination either form resolves to.

A queue that a subscription of this service binds takes every message pushed into it, whatever its name frame says. A reply published there would come back to the same handler as its next job. So the publisher refuses it with ZmqError::Send, naming the destination, and the in-process test broker refuses it the same way. The worker sends its replies through a second broker instead: an endpoint the sink binds, which the worker dials. bindable() turns that broker into a token, .bind(Publish) makes the reply policy publish through it, and register_broker adds it to the service.

On ZmqFanout a reply on the fan-out the subscription reads reaches every subscription of the service whose name is a prefix of the reply's name, its own included when it matches. On ZmqRpc a reply goes back to the peer that asked, as Responders shows.

Pipelines with PUSH/PULL sends a worker's results to the sink this way.

Example

The worker binds tcp://0.0.0.0:5555 for its jobs and dials the sink at tcp://sink:5556 for its results.

Macros
Manual

examples/zmq_publishing_replies.rs
/// The result names its own destination, so the subscriber clause stays bare.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    tracing::info!(job.id, "working");
    Done { id: job.id }
}

examples/zmq_publishing_replies.rs
#[ruststream::app]
fn app() -> impl App {
    // The sink binds its own endpoint; this worker dials it to deliver the results.
    let results = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let to_results = results.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(work).out_reply(to_results);
            },
        )
        .register_broker(results)
}

examples/manual/zmq_publishing_replies.rs
/// The result names its own destination, so the subscriber takes no name for its reply.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Done {
    id: u64,
}

// `#[outgoing(name = "results")]` by hand.
impl OutgoingDestination for Done {
    type Form = FixedName;
    const DESTINATION: &'static str = "results";
}

impl MessageHeaders for Done {
    type Contract = NoHeaders;
}

struct Work;

impl Handle<Job, Done> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Done, HandlerOutcome>> {
        tracing::info!(job.id, "working");
        ready(Ok(Done { id: job.id }))
    }
}

examples/manual/zmq_publishing_replies.rs
fn app() -> impl App {
    // The sink binds its own endpoint; this worker dials it to deliver the results.
    let results = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let to_results = results.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(subscriber("jobs", Work).reply().build())
                    .out_reply(to_results);
            },
        )
        .register_broker(results)
}

Example​

Example