Build an asynchronous server with DEALER and ROUTER

For Rust developers who have tried rzmq's PUSH/PULL and want request/reply without the lockstep

You will build a server on one ROUTER socket that answers several DEALER clients at once. Each client keeps a handful of requests in flight and matches the replies by request id as they come back out of order. Along the way you will see how the ROUTER names each client, what happens to a reply for a client that has already left and how to turn that into a visible error with ROUTER_MANDATORY.

Before you start

You need Rust with Cargo. rzmq comes from crates.io:

rzmq = "0"

Set up the project#

Start a new binary crate.

cargo new dealer-router
cd dealer-router

Add rzmq and Tokio to Cargo.toml:

Cargo.toml

[dependencies]
rzmq = "0"
tokio = { version = "1", features = ["full"] }

The default features cover everything here. The server and its clients all run in one process as Tokio tasks and talk TCP on 127.0.0.1:5650, so you can move any of them into its own process later without touching the socket code.

Serve two clients from one ROUTER#

REQ/REP makes a client wait for each reply before it can send the next request. That is fine for one quick call at a time, but a client with several independent jobs would rather send them all and take the answers as they finish. DEALER and ROUTER let you do that.

A DEALER can send any number of messages without waiting for replies. A ROUTER receives from every connected peer and puts one extra frame in front of each message: the identity of the peer it came from. To reply, you send a message whose first frame is that identity. The ROUTER strips it off and hands the rest to that one peer. The DEALER never sees the identity frame.

Because replies can come back in any order, each request carries its own id. In this program a request is two frames, the id and how many milliseconds of work it needs. The server sleeps that long to stand in for real work and then replies with the id and done. Every request gets its own Tokio task on the server, so a slow job doesn't hold up a quick one behind it.

Replace src/main.rs with this:

src/main.rs

use rzmq::socket::{ROUTING_ID, SocketEvent};
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use std::collections::HashMap;
use std::time::Duration;
use tokio::task::JoinSet;

const ENDPOINT: &str = "tcp://127.0.0.1:5650";
const CLIENTS: [(&str, &[u64]); 2] = [("alpha", &[60, 20, 40]), ("beta", &[50, 10, 30])];

fn text(msg: &Msg) -> String {
    String::from_utf8_lossy(msg.data().unwrap_or_default()).into_owned()
}

fn frame(value: impl ToString) -> Msg {
    Msg::from_vec(value.to_string().into_bytes())
}

async fn serve(router: Socket, requests: usize) -> Result<(), ZmqError> {
    let mut jobs = JoinSet::new();
    for _ in 0..requests {
        let frames = router.recv_multipart().await?;
        let (peer, id) = (frames[0].clone(), frames[1].clone());
        let ms: u64 = text(&frames[2]).parse().expect("work is a number");

        let router = router.clone();
        jobs.spawn(async move {
            tokio::time::sleep(Duration::from_millis(ms)).await;
            router.send_multipart(vec![peer, id, frame("done")]).await
        });
    }
    while let Some(sent) = jobs.join_next().await {
        sent.expect("reply task panicked")?;
    }
    Ok(())
}

async fn client(name: &str, dealer: Socket, work: &[u64]) -> Result<(), ZmqError> {
    let mut pending = HashMap::new();
    for (id, ms) in (1u64..).zip(work) {
        dealer.send_multipart(vec![frame(id), frame(ms)]).await?;
        pending.insert(id, *ms);
    }

    while !pending.is_empty() {
        let reply = dealer.recv_multipart().await?;
        let id: u64 = text(&reply[0]).parse().expect("request id is a number");
        let ms = pending.remove(&id).expect("reply matches a pending request");
        println!("{name}: #{id} {} after {ms} ms, {} still pending", text(&reply[1]), pending.len());
    }
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let router = ctx.socket(SocketType::Router)?;
    let events = router.monitor_default().await?;
    router.bind(ENDPOINT).await?;
    let requests = CLIENTS.iter().map(|(_, work)| work.len()).sum();
    let serving = tokio::spawn(serve(router, requests));

    let mut dealers = Vec::new();
    for (name, work) in CLIENTS {
        let dealer = ctx.socket(SocketType::Dealer)?;
        dealer.set_option(ROUTING_ID, name).await?;
        dealer.connect(ENDPOINT).await?;
        dealers.push((name, work, dealer));
    }

    let mut ready = 0;
    while let Ok(event) = events.recv().await {
        if let SocketEvent::HandshakeSucceeded { .. } = event {
            ready += 1;
            if ready == CLIENTS.len() {
                break;
            }
        }
    }

    let mut clients = JoinSet::new();
    for (name, work, dealer) in dealers {
        clients.spawn(client(name, dealer, work));
    }
    while let Some(done) = clients.join_next().await {
        done.expect("client panicked")?;
    }

    serving.await.expect("server panicked")?;
    ctx.term().await?;
    Ok(())
}

The frames, the reply tasks and the clients work like this:

The ROUTER receives three frames per request (identity, id and work) while the DEALER only sent two. The reply goes out as [peer, id, "done"] and arrives at the DEALER as [id, "done"].
Socket handles are cheap to clone. Each reply task gets its own clone of the ROUTER and they all send through the same socket.
Each client keeps a HashMap of pending requests. When a reply arrives it looks the id up and removes it, so it always knows which job just finished and how many are left.
ROUTING_ID gives each DEALER a name before it connects. That name is the identity frame the ROUTER sees.
As in the PUSH/PULL tutorial, main waits on the ROUTER's monitor for one HandshakeSucceeded per client before the clients start sending.

Run it:

$ cargo run
beta: #2 done after 10 ms, 2 still pending
alpha: #2 done after 20 ms, 2 still pending
beta: #3 done after 30 ms, 1 still pending
alpha: #3 done after 40 ms, 1 still pending
beta: #1 done after 50 ms, 0 still pending
alpha: #1 done after 60 ms, 0 still pending

Both clients sent all three requests up front. The replies come back in the order the work finishes: beta's second request (10 ms) first, alpha's first request (60 ms) last. The program uses Tokio's current_thread runtime and the work times are 10 ms apart, so you should see these lines in this order every time.

Let the ROUTER name a client#

A DEALER doesn't have to pick its own name. If you leave ROUTING_ID unset, the ROUTER makes one up when the peer connects. To see both kinds side by side, give beta no name and have the server print the identity frame of every reply it sends.

Change CLIENTS so each entry has an optional routing id:

src/main.rs

const CLIENTS: [(&str, Option<&str>, &[u64]); 2] = [
    ("alpha", Some("alpha"), &[60, 20, 40]),
    ("beta", None, &[50, 10, 30]),
];

Print the identity in the server's reply task, just before the send:

src/main.rs

        jobs.spawn(async move {
            tokio::time::sleep(Duration::from_millis(ms)).await;
            println!("server: reply #{} to {:?}", text(&id), text(&peer));
            router.send_multipart(vec![peer, id, frame("done")]).await
        });

In main, set ROUTING_ID only when there is one:

src/main.rs

    let requests = CLIENTS.iter().map(|(_, _, work)| work.len()).sum();
    let serving = tokio::spawn(serve(router, requests));

    let mut dealers = Vec::new();
    for (name, routing_id, work) in CLIENTS {
        let dealer = ctx.socket(SocketType::Dealer)?;
        if let Some(routing_id) = routing_id {
            dealer.set_option(ROUTING_ID, routing_id).await?;
        }
        dealer.connect(ENDPOINT).await?;
        dealers.push((name, work, dealer));
    }

Run it:

$ cargo run
server: reply #2 to "pipe:13"
beta: #2 done after 10 ms, 2 still pending
server: reply #2 to "alpha"
alpha: #2 done after 20 ms, 2 still pending
server: reply #3 to "pipe:13"
beta: #3 done after 30 ms, 1 still pending
server: reply #3 to "alpha"
alpha: #3 done after 40 ms, 1 still pending
server: reply #1 to "pipe:13"
beta: #1 done after 50 ms, 0 still pending
server: reply #1 to "alpha"
alpha: #1 done after 60 ms, 0 still pending

The generated name on your machine can be different from pipe:13 and it can change from run to run. Treat it as an opaque value: copy the frame you received into the reply and don't parse it or keep it after the connection is gone. Everything else in the output is fixed.

Set ROUTING_ID yourself when something outside the connection needs to know who a client is, for example a server that keeps per-client stats. Leave it unset when the server only ever answers what it was asked.

Reply to a client that has left#

A real server will sometimes finish a job for a client that is no longer there. Add a third client, gamma, that sends a 5 ms job and a 100 ms job, waits for the first reply and then closes its socket without waiting for the second.

The clients now differ in more ways, so this step describes them with a small struct. Each one also says how many replies it waits for and every client closes its DEALER when it is done. Replace src/main.rs:

src/main.rs

use rzmq::socket::{ROUTING_ID, SocketEvent};
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use std::collections::HashMap;
use std::time::Duration;
use tokio::task::JoinSet;

const ENDPOINT: &str = "tcp://127.0.0.1:5650";

struct Client {
    name: &'static str,
    routing_id: Option<&'static str>,
    work: &'static [u64],
    waits_for: usize,
}

const CLIENTS: [Client; 3] = [
    Client { name: "alpha", routing_id: Some("alpha"), work: &[60, 20, 40], waits_for: 3 },
    Client { name: "beta", routing_id: None, work: &[50, 10, 30], waits_for: 3 },
    Client { name: "gamma", routing_id: Some("gamma"), work: &[5, 100], waits_for: 1 },
];

fn text(msg: &Msg) -> String {
    String::from_utf8_lossy(msg.data().unwrap_or_default()).into_owned()
}

fn frame(value: impl ToString) -> Msg {
    Msg::from_vec(value.to_string().into_bytes())
}

async fn serve(router: Socket, requests: usize) -> Result<(), ZmqError> {
    let mut jobs = JoinSet::new();
    for _ in 0..requests {
        let frames = router.recv_multipart().await?;
        let (peer, id) = (frames[0].clone(), frames[1].clone());
        let ms: u64 = text(&frames[2]).parse().expect("work is a number");

        let router = router.clone();
        jobs.spawn(async move {
            tokio::time::sleep(Duration::from_millis(ms)).await;
            println!("server: reply #{} to {:?}", text(&id), text(&peer));
            router.send_multipart(vec![peer, id, frame("done")]).await
        });
    }
    while let Some(sent) = jobs.join_next().await {
        sent.expect("reply task panicked")?;
    }
    Ok(())
}

async fn client(spec: &Client, dealer: Socket) -> Result<(), ZmqError> {
    let mut pending = HashMap::new();
    for (id, ms) in (1u64..).zip(spec.work) {
        dealer.send_multipart(vec![frame(id), frame(ms)]).await?;
        pending.insert(id, *ms);
    }

    for _ in 0..spec.waits_for {
        let reply = dealer.recv_multipart().await?;
        let id: u64 = text(&reply[0]).parse().expect("request id is a number");
        let ms = pending.remove(&id).expect("reply matches a pending request");
        println!("{}: #{id} {} after {ms} ms, {} still pending", spec.name, text(&reply[1]), pending.len());
    }
    if !pending.is_empty() {
        println!("{}: leaving without waiting", spec.name);
    }
    dealer.close().await
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let router = ctx.socket(SocketType::Router)?;
    let events = router.monitor_default().await?;
    router.bind(ENDPOINT).await?;
    let requests = CLIENTS.iter().map(|spec| spec.work.len()).sum();
    let serving = tokio::spawn(serve(router, requests));

    let mut dealers = Vec::new();
    for spec in &CLIENTS {
        let dealer = ctx.socket(SocketType::Dealer)?;
        if let Some(routing_id) = spec.routing_id {
            dealer.set_option(ROUTING_ID, routing_id).await?;
        }
        dealer.connect(ENDPOINT).await?;
        dealers.push((spec, dealer));
    }

    let mut ready = 0;
    while let Ok(event) = events.recv().await {
        if let SocketEvent::HandshakeSucceeded { .. } = event {
            ready += 1;
            if ready == CLIENTS.len() {
                break;
            }
        }
    }

    let mut clients = JoinSet::new();
    for (spec, dealer) in dealers {
        clients.spawn(client(spec, dealer));
    }
    while let Some(done) = clients.join_next().await {
        done.expect("client panicked")?;
    }

    serving.await.expect("server panicked")?;
    ctx.term().await?;
    Ok(())
}

Gamma sends both requests before it waits and they travel in order over one connection. By the time its 5 ms reply arrives, the server has already read the 100 ms request, so the server still gets all eight requests it is counting on.

Run it:

$ cargo run
server: reply #1 to "gamma"
gamma: #1 done after 5 ms, 1 still pending
gamma: leaving without waiting
server: reply #2 to "pipe:15"
beta: #2 done after 10 ms, 2 still pending
server: reply #2 to "alpha"
alpha: #2 done after 20 ms, 2 still pending
server: reply #3 to "pipe:15"
beta: #3 done after 30 ms, 1 still pending
server: reply #3 to "alpha"
alpha: #3 done after 40 ms, 1 still pending
server: reply #1 to "pipe:15"
beta: #1 done after 50 ms, 0 still pending
server: reply #1 to "alpha"
alpha: #1 done after 60 ms, 0 still pending
server: reply #2 to "gamma"

The last line says the server replied to gamma, but nobody received that reply. With default options, a ROUTER asked to send to an identity it has no connection for drops the message and the send still returns Ok(()). A departed client can't block or fail the server that way, but the server also can't tell that the reply went nowhere.

Turn on ROUTER_MANDATORY#

The ROUTER_MANDATORY option changes that. With it set, sending to an identity the ROUTER doesn't know fails with ZmqError::HostUnreachable instead of being dropped quietly. Gamma's identity counts as unknown once its connection has gone.

Import the option alongside the others:

src/main.rs

use rzmq::socket::{ROUTER_MANDATORY, ROUTING_ID, SocketEvent};

Set it on the ROUTER before it binds:

src/main.rs

    let router = ctx.socket(SocketType::Router)?;
    router.set_option(ROUTER_MANDATORY, true).await?;
    let events = router.monitor_default().await?;
    router.bind(ENDPOINT).await?;

Then have the reply task handle the new error. A client that has left is a normal event for this server, so it logs it and moves on. Any other error still ends the server through ?.

src/main.rs

        jobs.spawn(async move {
            tokio::time::sleep(Duration::from_millis(ms)).await;
            let (who, n) = (text(&peer), text(&id));
            match router.send_multipart(vec![peer, id, frame("done")]).await {
                Ok(()) => println!("server: reply #{n} to {who:?}"),
                Err(ZmqError::HostUnreachable(_)) => println!("server: {who:?} is gone, dropped reply #{n}"),
                Err(e) => return Err(e),
            }
            Ok(())
        });

Run it:

$ cargo run
server: reply #1 to "gamma"
gamma: #1 done after 5 ms, 1 still pending
gamma: leaving without waiting
server: reply #2 to "pipe:16"
beta: #2 done after 10 ms, 2 still pending
server: reply #2 to "alpha"
alpha: #2 done after 20 ms, 2 still pending
server: reply #3 to "pipe:16"
beta: #3 done after 30 ms, 1 still pending
server: reply #3 to "alpha"
alpha: #3 done after 40 ms, 1 still pending
server: reply #1 to "pipe:16"
beta: #1 done after 50 ms, 0 still pending
server: reply #1 to "alpha"
alpha: #1 done after 60 ms, 0 still pending
server: "gamma" is gone, dropped reply #2

The server now knows reply #2 for gamma was not delivered. In a real service this is where you would cancel the rest of that client's work. The generated name for beta can differ on your machine, as before.

Shut down cleanly#

Every piece of this program knows when it is finished, so shutdown is a matter of waiting in the right order:

src/main.rs

    while let Some(done) = clients.join_next().await {
        done.expect("client panicked")?;
    }

    serving.await.expect("server panicked")?;
    ctx.term().await?;

Each client closes its own DEALER with close() after its last reply. The server's receive loop ends after it has read all eight requests. serve returns once every reply task has finished, including the one that found gamma gone. main waits for the clients, then the server, then calls Context::term(). That closes the ROUTER and anything else made from the context and waits for their background tasks before the program exits.

This server knows its workload up front. A long-running server doesn't, so it would run its receive loop inside tokio::select! next to a stop signal such as a tokio::sync::watch channel, wait for its reply tasks to drain and then call ctx.term().

Where to go next#

A client that can't wait forever for a reply can set RCVTIMEO in milliseconds. A recv that runs past it returns ZmqError::Timeout and the client can give up on whatever is still in its pending map.

src/main.rs

use rzmq::socket::RCVTIMEO;

dealer.set_option(RCVTIMEO, 500i32).await?;

SNDHWM and RCVHWM cap how many messages queue up per peer. The rzmq usage guide covers every socket option, the monitor events, security (PLAIN, CURVE and Noise_XX) and the io_uring backend.

The full API reference is on docs.rs.