# Build a log shipping pipeline with PUSH and PULL

**For:** Rust developers who know Tokio and want to see how PUSH/PULL handles fan-in, pipeline stages and slow consumers.

You will build a small log shipper over TCP on localhost. Three apps push their log lines into one collector, which stamps each line and pushes it on to a writer stage. Then a stalled collector shows what the high-water marks and a send timeout do when the next stage can't keep up.

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 logship
cd logship
```

Add rzmq and Tokio. rzmq runs on the Tokio runtime, so you need both. Your `[dependencies]` section should look like this:

```toml title="Cargo.toml"
[dependencies]
rzmq = "0"
tokio = { version = "1", features = ["full"] }
```

The default features cover everything here. Every socket talks TCP on `127.0.0.1`, so each piece could run as its own process. They all live in one process as Tokio tasks so you can run the whole thing at once. The programs use Tokio's single-threaded `current_thread` runtime, which keeps the output in the same order from run to run.

## Collect lines from several apps

The task ventilator in the earlier tutorial used PUSH/PULL to fan work out: one PUSH that binds, several PULL workers that connect. The same pair also works the other way round. Here one PULL socket is the stable end and binds. Each app connects its own PUSH socket to it. That is fan-in: many producers feed one consumer and none of the producers needs to know about the others.

A PULL socket only receives and a PUSH socket only sends, so there is no reply to wait for. An app hands a line to its socket and carries on.

Replace `src/main.rs` with this:

```rust title="src/main.rs"
use rzmq::{Context, Msg, SocketType, ZmqError};

const COLLECTOR: &str = "tcp://127.0.0.1:5640";
const APPS: [&str; 3] = ["api", "auth", "billing"];
const LINES_PER_APP: usize = 3;

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

    let collector = ctx.socket(SocketType::Pull)?;
    collector.bind(COLLECTOR).await?;

    let mut apps = Vec::new();
    for app in APPS {
        let shipper = ctx.socket(SocketType::Push)?;
        shipper.connect(COLLECTOR).await?;
        apps.push(tokio::spawn(async move {
            for n in 1..=LINES_PER_APP {
                let line = format!("{app}: request {n} done");
                shipper.send(Msg::from_vec(line.into_bytes())).await?;
            }
            Ok::<(), ZmqError>(())
        }));
    }

    for _ in 0..APPS.len() * LINES_PER_APP {
        let msg = collector.recv().await?;
        println!("{}", String::from_utf8_lossy(msg.data().unwrap_or_default()));
    }

    for app in apps {
        app.await.expect("app task panicked")?;
    }
    ctx.term().await?;
    Ok(())
}
```

Each app gets its own PUSH socket and runs in its own Tokio task. The collector knows that three apps send three lines each, so it reads nine messages and stops. `main` then waits for the app tasks and calls `ctx.term()`, which closes every socket made from the context.

Run it:

```shell-session
$ cargo run
api: request 1 done
api: request 2 done
api: request 3 done
auth: request 1 done
auth: request 2 done
auth: request 3 done
billing: request 1 done
billing: request 2 done
billing: request 3 done
```

The lines from one app always arrive in the order that app sent them. How lines from different apps mix depends on when each app sends. On the single-threaded runtime each app task sends its three lines in one go, so they arrive grouped by app.

## Add a writer stage

A real collector does something with each line and hands it on. Pipelines are what PUSH/PULL is for: each stage reads from a PULL socket, does its work and writes to a PUSH socket that the next stage reads from. Here the collector becomes an aggregator that gives every line a sequence number and pushes it to a writer.

The collector can no longer count on a fixed number of lines, so the apps say when they are done. Each app sends an `END` message after its last line. The aggregator counts the `END` messages and, once all three apps are finished, sends one `END` of its own down to the writer. The writer prints lines until it sees that `END`. A PUSH socket keeps each sender's messages in order, so an app's `END` always arrives after its last line.

The apps now start before the aggregator binds. That is the missing-peer case: a PUSH socket with nothing connected to it waits in `send` until a peer is there. Meanwhile the `connect` keeps retrying in the background until the aggregator's `bind` shows up. Nothing is lost and no app needs to know whether the collector is up yet.

Replace `src/main.rs` with the full program:

```rust title="src/main.rs"
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};

const COLLECTOR: &str = "tcp://127.0.0.1:5640";
const WRITER: &str = "tcp://127.0.0.1:5641";
const APPS: [&str; 3] = ["api", "auth", "billing"];
const LINES_PER_APP: usize = 3;
const END: &[u8] = b"END";

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

async fn app(name: &'static str, shipper: Socket) -> Result<(), ZmqError> {
    for n in 1..=LINES_PER_APP {
        let line = format!("{name}: request {n} done");
        shipper.send(Msg::from_vec(line.into_bytes())).await?;
    }
    shipper.send(Msg::from_static(END)).await
}

async fn aggregator(inbox: Socket, outbox: Socket) -> Result<(), ZmqError> {
    let mut seq = 0;
    let mut finished = 0;
    while finished < APPS.len() {
        let msg = inbox.recv().await?;
        if msg.data() == Some(END) {
            finished += 1;
            continue;
        }
        seq += 1;
        let stamped = format!("{seq:04} {}", text(&msg));
        outbox.send(Msg::from_vec(stamped.into_bytes())).await?;
    }
    outbox.send(Msg::from_static(END)).await
}

async fn writer(inbox: Socket) -> Result<usize, ZmqError> {
    let mut written = 0;
    loop {
        let msg = inbox.recv().await?;
        if msg.data() == Some(END) {
            return Ok(written);
        }
        println!("{}", text(&msg));
        written += 1;
    }
}

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

    let mut apps = Vec::new();
    for name in APPS {
        let shipper = ctx.socket(SocketType::Push)?;
        shipper.connect(COLLECTOR).await?;
        apps.push(tokio::spawn(app(name, shipper)));
    }

    let inbox = ctx.socket(SocketType::Pull)?;
    inbox.bind(COLLECTOR).await?;
    let outbox = ctx.socket(SocketType::Push)?;
    outbox.bind(WRITER).await?;
    let aggregating = tokio::spawn(aggregator(inbox, outbox));

    let writer_inbox = ctx.socket(SocketType::Pull)?;
    writer_inbox.connect(WRITER).await?;
    let written = writer(writer_inbox).await?;
    println!("writer: {written} lines");

    aggregating.await.expect("aggregator panicked")?;
    for app in apps {
        app.await.expect("app panicked")?;
    }
    ctx.term().await?;
    Ok(())
}
```

The aggregator binds both of its sockets. It sits in the middle of the pipeline and is the piece least likely to move, so the apps and the writer connect to it. `writer` runs on the main task and returns the number of lines it wrote.

Run it:

```shell-session
$ cargo run
0001 auth: request 1 done
0002 auth: request 2 done
0003 auth: request 3 done
0004 api: request 1 done
0005 api: request 2 done
0006 api: request 3 done
0007 billing: request 1 done
0008 billing: request 2 done
0009 billing: request 3 done
writer: 9 lines
```

The apps have to reconnect once the aggregator binds. The order they get through that can change between runs, so the apps may appear in a different order. The sequence numbers always run from 0001 to 0009, each app's three lines stay in order and the writer always reports nine lines.

## Stall the collector

So far every stage keeps up. When one doesn't, messages queue up. Two socket options put a limit on that queue. `SNDHWM` is the high-water mark for the sending side: how many messages a PUSH socket holds for a peer that hasn't taken them yet. `RCVHWM` is the same for the receiving side: how many a PULL socket holds that your code hasn't read with `recv`. When the queue is full, `send` on the PUSH socket waits for space. That is the backpressure in this pattern. A slow consumer slows its producers down instead of making them buffer without end. The socket doesn't drop any message.

Waiting is right for the aggregator, which has nothing better to do. An app shouldn't stop serving requests because its log collector is stuck. `SNDTIMEO` sets how long `send` may wait, in milliseconds. When the time runs out, `send` returns `ZmqError::Timeout` and the message was not sent. The app can then decide what to do. Here it drops the rest of its burst.

This step uses one app talking straight to a collector, so there is only one queue to watch. The collector is stalled: it reads nothing until the app tells it how many lines got through. The app tries to ship a burst of 100,000 lines.

Replace `src/main.rs`:

```rust title="src/main.rs"
use rzmq::socket::{RCVHWM, SNDHWM, SNDTIMEO};
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use tokio::sync::oneshot;

const COLLECTOR: &str = "tcp://127.0.0.1:5642";
const HWM: i32 = 100;
const BURST: usize = 100_000;

async fn stalled_collector(inbox: Socket, resume: oneshot::Receiver<usize>) -> Result<(), ZmqError> {
    let shipped = resume.await.expect("app went away");
    for _ in 0..shipped {
        inbox.recv().await?;
    }
    println!("collector: caught up on {shipped} lines");
    Ok(())
}

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

    let inbox = ctx.socket(SocketType::Pull)?;
    inbox.set_option(RCVHWM, HWM).await?;
    inbox.bind(COLLECTOR).await?;
    let (resume, resumed) = oneshot::channel();
    let collecting = tokio::spawn(stalled_collector(inbox, resumed));

    let shipper = ctx.socket(SocketType::Push)?;
    shipper.set_option(SNDHWM, HWM).await?;
    shipper.set_option(SNDTIMEO, 100i32).await?;
    shipper.connect(COLLECTOR).await?;

    let mut shipped = 0;
    for n in 1..=BURST {
        let line = format!("api: request {n} done");
        match shipper.send(Msg::from_vec(line.into_bytes())).await {
            Ok(()) => shipped += 1,
            Err(ZmqError::Timeout) => break,
            Err(e) => return Err(e),
        }
    }
    println!("app: shipped {shipped} lines, dropped {}", BURST - shipped);

    resume.send(shipped).expect("collector went away");
    collecting.await.expect("collector panicked")?;
    ctx.term().await?;
    Ok(())
}
```

Both options take an `i32` and are set before `bind` and `connect`. The high-water marks go on both ends, `SNDHWM` on the PUSH and `RCVHWM` on the PULL. The `oneshot` channel stands in for whatever unsticks a real collector. It also tells the collector how many lines to read.

Run it:

```shell-session
$ cargo run
app: shipped 22800 lines, dropped 77200
collector: caught up on 22800 lines
```

Your numbers will likely differ. Messages also sit in the operating system's TCP buffers between the two sockets, so the app gets further than the two high-water marks alone would allow. How much further depends on the machine and the size of each line. What stays fixed is that shipped plus dropped is always 100,000 and the collector gets every line that was shipped. Raise `HWM` and more lines get through before the timeout; lower it and fewer do.

Without `SNDTIMEO` the app's first `send` into a full queue would wait until the collector started reading again. That is exactly what you want for a pipeline stage like the aggregator, where a slow writer should slow down everything upstream of it.

## Shut down cleanly

In the pipeline every task knows when it is finished, because the `END` messages travel through every stage. Shutdown is waiting for the tasks in order and then terminating the context:

```rust title="src/main.rs"
let writer_inbox = ctx.socket(SocketType::Pull)?;
writer_inbox.connect(WRITER).await?;
let written = writer(writer_inbox).await?;
println!("writer: {written} lines");

aggregating.await.expect("aggregator panicked")?;
for app in apps {
    app.await.expect("app panicked")?;
}
ctx.term().await?;
```

The writer returns after the aggregator's `END`, which the aggregator only sends after the `END` from every app. By the time `main` calls `Context::term()`, every message has been read and no task still holds a socket it is using. `term()` closes every socket made from the context and waits for their background tasks to finish. Messages still queued at that point are handled according to each socket's `LINGER` option. Any error along the way is passed up with `?` instead of being ignored.

The `END` marker is one message per producer. The stage in the middle has to count them before passing its own on, since a PULL socket has no idea how many producers are connected to it.

## Where to go next

Every stage in this tutorial is a PULL that reads and a PUSH that writes, so you can add stages or run several copies of one. Several writers connected to the aggregator's PUSH would share the lines round-robin, which is the fan-out shape from the ventilator tutorial. Other options work the same way as the ones here. `RCVTIMEO` puts a time limit on `recv` and returns `ZmqError::Timeout` when it runs out. `monitor_default()` reports `ConnectionCongested` and `ConnectionUncongested` events when a peer's send queue hits its high-water mark and drains again.

The [rzmq usage guide](https://github.com/excsn/rzmq/blob/main/core/README.USAGE.md) on GitHub covers socket options, monitoring, security and the io_uring backend.

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