Wire format

Every message is three frames, name, headers, payload, so a peer in any language composes one with send_multipart.

ZeroMQ carries messages made of frames and leaves their meaning to the application. This crate gives every message on PUSH/PULL and PUB/SUB the same three frames: the name, the headers and the payload.

The name frame is the subscription's name on the fan-out, matched as a prefix; a queue takes every message whatever its name says. The header frame is text, and an empty frame means no headers. A message of two frames, a name and a payload, reads as a message without headers, so a minimal peer can skip the middle frame. The payload frame is what the service's codec wrote, so a peer agrees on the codec only: with the default JSON codec it is the document the handler's input type deserializes from. Bytes a service already holds encoded go out through a #[derive(Outgoing, Serialized)] newtype, and no codec runs on them. A name frame that is not UTF-8 surfaces on the subscription as ZmqError::Wire.

On DEALER/ROUTER a request has the same three frames, and its answer travels behind the requester's identity, as Envelope shows.

Peers in other languages feeds, watches and asks a service from Python peers.

Example

frame 0: name      UTF-8; the fan-out's subscribers filter on it as a prefix
frame 1: headers   UTF-8 "name: value" lines separated by \n; may be empty
frame 2: payload   the codec's output

A Python peer pushes a job into a worker with pyzmq:

socket.send_multipart([b"jobs", b"content-type: application/json", b'{"id": 2}'])

The worker reads the jobs a Python PUSH socket sends, and pushes each result to a sink that a Python PULL socket binds. The sink reads the three frames results, the header lines and the JSON document {"id":2}.

Macros
Manual

examples/zmq_wire.rs
/// What the peer's payload frame holds: the JSON document this type deserializes from.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "jobs")]
struct Job {
    id: u64,
}

#[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_wire.rs
fn service(jobs: ZmqEndpoint, sink: ZmqEndpoint<Connect>) -> impl App {
    let results = ZmqQueue::new(sink).bindable();
    let to_results = results.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(jobs), |b| {
            b.include(work).out_reply(to_results);
        })
        .register_broker(results)
}

#[ruststream::app]
fn app() -> impl App {
    service(
        ZmqEndpoint::bind("tcp://0.0.0.0:5555"),
        ZmqEndpoint::connect("tcp://sink:5556"),
    )
}

examples/manual/zmq_wire.rs
/// What the peer's payload frame holds: the JSON document this type deserializes from.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Job {
    id: u64,
}

// `#[outgoing(name = "jobs")]` by hand: the type names the subscription it is pushed to.
impl OutgoingDestination for Job {
    type Form = FixedName;
    const DESTINATION: &'static str = "jobs";
}

impl MessageHeaders for Job {
    type Contract = NoHeaders;
}

#[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_wire.rs
fn service(jobs: ZmqEndpoint, sink: ZmqEndpoint<Connect>) -> impl App {
    let results = ZmqQueue::new(sink).bindable();
    let to_results = results.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(jobs), |b| {
            b.include(subscriber("jobs", Work).reply().build())
                .out_reply(to_results);
        })
        .register_broker(results)
}

fn app() -> impl App {
    service(
        ZmqEndpoint::bind("tcp://0.0.0.0:5555"),
        ZmqEndpoint::connect("tcp://sink:5556"),
    )
}

Example​

Example