# 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 as `rzmq = "0"`.

## Set up the project

Start a new binary crate.

```sh
cargo new lookup
cd lookup
```

Add rzmq and Tokio to `Cargo.toml`:

```toml title="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`:

```rust title="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:

```shell-session
$ 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`:

```rust title="src/main.rs" {1,3,19-22,25-28,30-32,34-35}
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:

```shell-session
$ 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.

## Share one server between clients

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`:

```rust title="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:

```shell-session
$ 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`:

```rust title="src/main.rs" {1,3,5-6,18,28-30,33,38-43,45-47,56,58-59,61-68,71}
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:

```shell-session
$ 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`:

```rust title="src/main.rs" {4,32,34-38,59,64,67,72,74-77,79-81,83}
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:

```shell-session
$ 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:

```rust title="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`:

```rust title="src/main.rs"
req.set_option(RCVTIMEO, TIMEOUT_MS).await?;
req.set_option(SNDTIMEO, TIMEOUT_MS).await?;
```

The [rzmq usage guide](https://github.com/excsn/rzmq/blob/main/core/README.USAGE.md) on GitHub covers socket options such as `RCVTIMEO`, `SNDTIMEO` and `LINGER`, security, monitoring and error handling.

The full API reference is on [docs.rs](https://docs.rs/rzmq).