Build a lookup service with REQ and REP

For Rust developers who know Tokio and want a simple request/reply service over ZeroMQ

You will build a small port lookup service over TCP on localhost. A REP socket answers questions like "what runs on port 22?" and REQ clients ask them. Along the way you will break the send/receive rule on purpose, share one server between three clients, give up on a slow reply with RCVTIMEO and recover when the server goes away.

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 lookup
cd lookup

Add rzmq and Tokio to Cargo.toml:

Cargo.toml

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

The default features cover everything here. Each step replaces src/main.rs with a complete program. The programs use Tokio's single-threaded runtime so the output comes out in the same order every time.

Answer lookups with REP#

REQ/REP is the plain client/server pattern. A REQ socket sends a request and then waits for the reply. A REP socket receives a request and then sends the reply. The server is the stable end, so the REP socket binds and the client connects to it.

Replace src/main.rs:

src/main.rs

use rzmq::{Context, Msg, Socket, SocketType, ZmqError};

const SERVICE: &str = "tcp://127.0.0.1:5620";

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

fn lookup(port: &str) -> &'static str {
    match port {
        "22" => "ssh",
        "53" => "dns",
        "80" => "http",
        "443" => "https",
        "5432" => "postgres",
        _ => "unknown",
    }
}

async fn serve(rep: Socket, requests: usize) -> Result<(), ZmqError> {
    for _ in 0..requests {
        let request = rep.recv().await?;
        let port = text(&request);
        let answer = format!("{port} {}", lookup(&port));
        rep.send(Msg::from_vec(answer.into_bytes())).await?;
    }
    Ok(())
}

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

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;
    let server = tokio::spawn(serve(rep, 3));

    let req = ctx.socket(SocketType::Req)?;
    req.connect(SERVICE).await?;

    for port in ["22", "443", "8080"] {
        req.send(Msg::from_vec(port.as_bytes().to_vec())).await?;
        let reply = req.recv().await?;
        println!("port {port}: {}", text(&reply));
    }

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

The server runs in its own Tokio task and answers a fixed number of requests before it returns. The client asks three questions, one at a time. Each send is followed by a recv that waits for the matching answer. Once the server task has finished, ctx.term() closes both sockets.

Run it:

$ cargo run
port 22: 22 ssh
port 443: 443 https
port 8080: 8080 unknown

Keep to send, then receive#

A REQ socket must alternate: send, receive, send, receive. A REP socket does the same the other way round: receive, then send. A call out of order returns ZmqError::InvalidState. This step breaks the order on purpose to show what that looks like.

Replace src/main.rs:

src/main.rs

use rzmq::{Context, Msg, SocketType, ZmqError};

const SERVICE: &str = "tcp://127.0.0.1:5621";

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

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

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;

    let req = ctx.socket(SocketType::Req)?;
    req.connect(SERVICE).await?;

    req.send(Msg::from_static(b"22")).await?;
    match req.send(Msg::from_static(b"53")).await {
        Err(e) => println!("REQ second send: {e:?}"),
        Ok(()) => println!("REQ second send was accepted"),
    }

    match rep.send(Msg::from_static(b"too early")).await {
        Err(e) => println!("REP send before recv: {e:?}"),
        Ok(()) => println!("REP send before recv was accepted"),
    }

    let request = rep.recv().await?;
    println!("REP got {}", text(&request));
    rep.send(Msg::from_static(b"22 ssh")).await?;

    let reply = req.recv().await?;
    println!("REQ got {}", text(&reply));

    ctx.term().await?;
    Ok(())
}

Run it:

$ cargo run
REQ second send: InvalidState("REQ socket must call recv() before sending again")
REP send before recv: InvalidState("REP socket must recv() a request before sending a reply")
REP got 22
REQ got 22 ssh

Both calls return the error and send nothing. The rejected 53 never reaches the server, which only sees 22. After the errors both sockets carry on normally. If you need several requests in flight at once, REQ/REP is the wrong pair; DEALER and ROUTER are the asynchronous versions.

A REP server can have many REQ clients connected at once. It takes their requests one at a time and each reply goes back to the client that asked. This step starts three clients, each in its own task with its own REQ socket. Each one asks two questions. The server now sleeps 50 ms per lookup to stand in for real work.

Replace src/main.rs:

src/main.rs

use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use std::time::{Duration, Instant};

const SERVICE: &str = "tcp://127.0.0.1:5622";
const CLIENTS: usize = 3;

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

fn lookup(port: &str) -> &'static str {
    match port {
        "22" => "ssh",
        "53" => "dns",
        "80" => "http",
        "443" => "https",
        "5432" => "postgres",
        _ => "unknown",
    }
}

async fn serve(rep: Socket, requests: usize) -> Result<(), ZmqError> {
    for _ in 0..requests {
        let request = rep.recv().await?;
        let port = text(&request);
        tokio::time::sleep(Duration::from_millis(50)).await;
        let answer = format!("{port} {}", lookup(&port));
        rep.send(Msg::from_vec(answer.into_bytes())).await?;
    }
    Ok(())
}

async fn client(req: Socket, ports: [&'static str; 2]) -> Result<Vec<String>, ZmqError> {
    let mut answers = Vec::new();
    for port in ports {
        req.send(Msg::from_vec(port.as_bytes().to_vec())).await?;
        answers.push(text(&req.recv().await?));
    }
    Ok(answers)
}

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

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;
    let server = tokio::spawn(serve(rep, CLIENTS * 2));

    let questions = [["22", "53"], ["80", "443"], ["5432", "8080"]];
    let started = Instant::now();
    let mut clients = Vec::new();
    for ports in questions {
        let req = ctx.socket(SocketType::Req)?;
        req.connect(SERVICE).await?;
        clients.push(tokio::spawn(client(req, ports)));
    }

    for (id, handle) in clients.into_iter().enumerate() {
        let answers = handle.await.expect("client panicked")?;
        println!("client-{}: {}", id + 1, answers.join(", "));
    }
    server.await.expect("server panicked")?;
    println!("6 lookups took {:.1}s", started.elapsed().as_secs_f64());

    ctx.term().await?;
    Ok(())
}

Each client task returns its answers and main prints them in client order. The order in which the server picks up the six requests can change from run to run, but every client gets the answers to its own questions.

Run it:

$ cargo run
client-1: 22 ssh, 53 dns
client-2: 80 http, 443 https
client-3: 5432 postgres, 8080 unknown
6 lookups took 0.3s

Six lookups at 50 ms each take about 0.3 seconds. A REP socket works on one request at a time, so a slow lookup holds up every client queued behind it. When one slow request shouldn't hold up the rest, move to DEALER/ROUTER.

Give up on a slow reply#

By default a REQ client waits in recv until the reply comes. The RCVTIMEO option puts a limit on that wait, in milliseconds. When it runs out, recv returns ZmqError::Timeout.

After a timeout the client throws its REQ socket away: it closes it with close() and opens a new one. That is the usual ZeroMQ recovery for a lost request. Socket setup moves into client_socket so the client can make a fresh one whenever it needs to.

In this step, looking up port 3306 takes 400 ms on the server and the client waits at most 250 ms. The server now prints a line each time it sends an answer.

Replace src/main.rs:

src/main.rs

use rzmq::socket::RCVTIMEO;
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use std::time::Duration;

const SERVICE: &str = "tcp://127.0.0.1:5623";
const TIMEOUT_MS: i32 = 250;

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

fn lookup(port: &str) -> &'static str {
    match port {
        "22" => "ssh",
        "53" => "dns",
        "80" => "http",
        "443" => "https",
        "3306" => "mysql",
        "5432" => "postgres",
        _ => "unknown",
    }
}

async fn serve(rep: Socket, requests: usize) -> Result<(), ZmqError> {
    for _ in 0..requests {
        let request = rep.recv().await?;
        let port = text(&request);
        if port == "3306" {
            tokio::time::sleep(Duration::from_millis(400)).await;
        }
        let answer = format!("{port} {}", lookup(&port));
        rep.send(Msg::from_vec(answer.into_bytes())).await?;
        println!("server: answered {port}");
    }
    Ok(())
}

async fn client_socket(ctx: &Context) -> Result<Socket, ZmqError> {
    let req = ctx.socket(SocketType::Req)?;
    req.set_option(RCVTIMEO, TIMEOUT_MS).await?;
    req.connect(SERVICE).await?;
    Ok(req)
}

async fn ask(req: &Socket, port: &str) -> Result<String, ZmqError> {
    req.send(Msg::from_vec(port.as_bytes().to_vec())).await?;
    Ok(text(&req.recv().await?))
}

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

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;
    let server = tokio::spawn(serve(rep, 3));

    let mut req = client_socket(&ctx).await?;
    println!("client: {}", ask(&req, "22").await?);

    match ask(&req, "3306").await {
        Ok(answer) => println!("client: {answer}"),
        Err(ZmqError::Timeout) => {
            println!("client: no reply for 3306 within {TIMEOUT_MS} ms");
            req.close().await?;
            req = client_socket(&ctx).await?;
        }
        Err(e) => return Err(e),
    }

    println!("client: {}", ask(&req, "443").await?);

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

Run it:

$ cargo run
server: answered 22
client: 22 ssh
client: no reply for 3306 within 250 ms
server: answered 3306
server: answered 443
client: 443 https

The server doesn't know the client gave up. It finishes the 3306 lookup, sends the answer toward a socket that has been closed and then takes the 443 request from the new socket. The new request waits behind the slow one, which is why the timeout has to be longer than the time left on the slow lookup.

Recover when the server goes away#

The same recovery covers a server that dies in the middle of a request. The first server here answers one lookup, receives the next one and then closes its socket without replying. The client's recv hits RCVTIMEO, the client closes its REQ socket, a second server binds the same endpoint and the client retries on a fresh socket.

Replace src/main.rs:

src/main.rs

use rzmq::socket::RCVTIMEO;
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};

const SERVICE: &str = "tcp://127.0.0.1:5624";
const TIMEOUT_MS: i32 = 250;

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

fn lookup(port: &str) -> &'static str {
    match port {
        "22" => "ssh",
        "53" => "dns",
        "80" => "http",
        "443" => "https",
        "3306" => "mysql",
        "5432" => "postgres",
        _ => "unknown",
    }
}

async fn serve(rep: Socket, requests: usize) -> Result<(), ZmqError> {
    for _ in 0..requests {
        let request = rep.recv().await?;
        let port = text(&request);
        let answer = format!("{port} {}", lookup(&port));
        rep.send(Msg::from_vec(answer.into_bytes())).await?;
        println!("server: answered {port}");
    }
    Ok(())
}

async fn crashing_server(rep: Socket) -> Result<(), ZmqError> {
    serve(rep.clone(), 1).await?;
    let request = rep.recv().await?;
    println!("server: got {} and crashed", text(&request));
    rep.close().await
}

async fn client_socket(ctx: &Context) -> Result<Socket, ZmqError> {
    let req = ctx.socket(SocketType::Req)?;
    req.set_option(RCVTIMEO, TIMEOUT_MS).await?;
    req.connect(SERVICE).await?;
    Ok(req)
}

async fn ask(req: &Socket, port: &str) -> Result<String, ZmqError> {
    req.send(Msg::from_vec(port.as_bytes().to_vec())).await?;
    Ok(text(&req.recv().await?))
}

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

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;
    let first_server = tokio::spawn(crashing_server(rep));

    let mut req = client_socket(&ctx).await?;
    println!("client: {}", ask(&req, "22").await?);

    match ask(&req, "53").await {
        Ok(answer) => println!("client: {answer}"),
        Err(ZmqError::Timeout) => {
            println!("client: no reply for 53 within {TIMEOUT_MS} ms");
            req.close().await?;
        }
        Err(e) => return Err(e),
    }
    first_server.await.expect("server panicked")?;

    let rep = ctx.socket(SocketType::Rep)?;
    rep.bind(SERVICE).await?;
    let second_server = tokio::spawn(serve(rep, 2));
    println!("server: restarted");

    req = client_socket(&ctx).await?;
    println!("client: retry gave {}", ask(&req, "53").await?);
    println!("client: {}", ask(&req, "80").await?);

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

Run it:

$ cargo run
server: answered 22
client: 22 ssh
server: got 53 and crashed
client: no reply for 53 within 250 ms
server: restarted
server: answered 53
client: retry gave 53 dns
server: answered 80
client: 80 http

A retry means the server may see the same request twice: the first server received 53 before it died and the second one answered it again. That is harmless for a lookup. For a request that changes something, such as charging a card, the server needs a way to spot a repeat, for example a request ID it has already handled.

Shut down cleanly#

Every server in this tutorial answers a known number of requests and then returns, so shutdown is a matter of waiting for it and then terminating the context:

src/main.rs

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

Context::term() closes every socket made from the context and waits for their background tasks to finish. That includes a REQ socket you replaced but never closed. Socket::close() shuts down one socket early, which is what the client uses to throw away a socket after a lost request. Errors go up with ? so a failed run exits with an error instead of hiding it.

Where to go next#

REQ/REP keeps one request in flight per client. When a client needs several requests outstanding or a server needs to answer out of order, use DEALER for the client and ROUTER for the server. ROUTER hands you the sender's identity as the first frame so you can route each reply yourself.

RCVTIMEO only limits recv. SNDTIMEO puts the same kind of limit on send. A send that runs out of time also returns ZmqError::Timeout. A client that should never hang sets both in client_socket:

src/main.rs

req.set_option(RCVTIMEO, TIMEOUT_MS).await?;
req.set_option(SNDTIMEO, TIMEOUT_MS).await?;

The rzmq usage guide on GitHub covers socket options such as RCVTIMEO, SNDTIMEO and LINGER, security, monitoring and error handling.

The full API reference is on docs.rs.

Set up the project#

Answer lookups with REP#

Keep to send, then receive#

Share one server between clients#

Give up on a slow reply#

Recover when the server goes away#

Shut down cleanly#

Where to go next#