# Build a price feed with PUB/SUB topic filtering

**For:** Rust developers who know Tokio and want to broadcast updates with ZeroMQ patterns without the C library.

You will build a small price feed over TCP on localhost. One PUB socket publishes stock and currency prices as two-frame messages (topic, then price) and several SUB sockets pick out the topics they care about by prefix. Along the way you change a subscription with `UNSUBSCRIBE`, see why updates published before a subscriber is ready never reach it and cap what a slow subscriber can hold up with `SNDHWM`.

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

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. All the sockets talk TCP on `127.0.0.1`, so the publisher and each subscriber could run in separate processes. In this tutorial they are Tokio tasks in one process, which keeps each step to one `cargo run`. The program uses Tokio's `current_thread` runtime so the output comes out in the same order every run.

## Publish to one subscriber

A PUB socket sends every message to all connected SUB sockets whose subscriptions match it. A SUB socket receives nothing until you give it at least one subscription with the `SUBSCRIBE` option. A subscription is a byte prefix: `"stock."` matches any message whose first frame starts with `stock.` and the empty string matches everything.

Each update here is two frames sent with `send_multipart`: the topic, then the price. Subscriptions are matched against the start of the first frame, so putting the topic in a frame of its own keeps the filter away from the payload.

PUB/SUB has no delivery guarantee for peers that aren't there yet. The publisher drops a message when no connected subscriber has a matching subscription. A new subscriber's subscriptions reach the publisher a moment after its connection is up. Waiting for the `HandshakeSucceeded` monitor event, as the task ventilator tutorial does for PUSH, isn't enough here: updates published right after that event can still be filtered out because the subscription hasn't arrived yet.

This program confirms readiness instead. The publisher repeats a `sys.hello` message every 10 ms. Each subscriber subscribes to `sys.hello` along with its real topics. When the first hello comes through, it tells the publisher over a separate PUSH/PULL channel. A hello can only get through once the subscriber's subscriptions have reached the publisher, so after one confirmation per subscriber the publisher knows everyone is listening. A `sys.end` message tells subscribers the run is over.

Replace `src/main.rs` with this:

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

const PRICES: &str = "tcp://127.0.0.1:5630";
const READY: &str = "tcp://127.0.0.1:5631";

const UPDATES: [(&str, &str); 3] = [
    ("stock.AAPL", "189.25"),
    ("fx.EURUSD", "1.0842"),
    ("stock.MSFT", "411.80"),
];

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

async fn publish(publisher: &Socket, topic: &str, body: &str) -> Result<(), ZmqError> {
    publisher
        .send_multipart(vec![
            Msg::from_vec(topic.as_bytes().to_vec()),
            Msg::from_vec(body.as_bytes().to_vec()),
        ])
        .await
}

async fn wait_for_subscribers(
    publisher: &Socket,
    ready: &Socket,
    count: usize,
    hello: &str,
) -> Result<(), ZmqError> {
    let ready = ready.clone();
    let confirmations = tokio::spawn(async move {
        for _ in 0..count {
            ready.recv().await?;
        }
        Ok::<(), ZmqError>(())
    });

    let mut tick = tokio::time::interval(Duration::from_millis(10));
    while !confirmations.is_finished() {
        tick.tick().await;
        publish(publisher, hello, "").await?;
    }
    confirmations.await.expect("confirmation task panicked")
}

async fn confirm(prices: &Socket, ready: &Socket, hello: &str) -> Result<(), ZmqError> {
    loop {
        let frames = prices.recv_multipart().await?;
        if text(&frames[0]) == hello {
            return ready.send(Msg::new()).await;
        }
    }
}

async fn read_round(prices: &Socket) -> Result<Vec<String>, ZmqError> {
    let mut seen = Vec::new();
    loop {
        let frames = prices.recv_multipart().await?;
        let topic = text(&frames[0]);
        if topic == "sys.end" {
            return Ok(seen);
        }
        if !topic.starts_with("sys.hello") {
            seen.push(format!("{topic} {}", text(&frames[1])));
        }
    }
}

async fn subscriber(ctx: Context, prefix: &'static str) -> Result<Vec<String>, ZmqError> {
    let prices = ctx.socket(SocketType::Sub)?;
    for topic in ["sys.hello", "sys.end", prefix] {
        prices.set_option(SUBSCRIBE, topic).await?;
    }
    prices.connect(PRICES).await?;

    let ready = ctx.socket(SocketType::Push)?;
    ready.connect(READY).await?;

    confirm(&prices, &ready, "sys.hello").await?;
    read_round(&prices).await
}

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

    let publisher = ctx.socket(SocketType::Pub)?;
    publisher.bind(PRICES).await?;
    let ready = ctx.socket(SocketType::Pull)?;
    ready.bind(READY).await?;

    publish(&publisher, "stock.AAPL", "189.10").await?;
    println!("published stock.AAPL 189.10 before anyone subscribed");

    let stocks = tokio::spawn(subscriber(ctx.clone(), "stock."));
    wait_for_subscribers(&publisher, &ready, 1, "sys.hello").await?;
    println!("subscriber is listening");

    for (topic, price) in UPDATES {
        publish(&publisher, topic, price).await?;
    }
    publish(&publisher, "sys.end", "").await?;

    for line in stocks.await.expect("subscriber panicked")? {
        println!("received {line}");
    }

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

`wait_for_subscribers` counts confirmations in a spawned task and keeps publishing hellos until that task finishes. `confirm` waits for the first hello and answers it. `read_round` collects updates until `sys.end` and skips any hellos still in flight. Before anyone connects, `main` publishes one AAPL price that nobody will ever see.

Run it:

```shell-session
$ cargo run
published stock.AAPL 189.10 before anyone subscribed
subscriber is listening
received stock.AAPL 189.25
received stock.MSFT 411.80
```

The 189.10 update is gone for good. PUB doesn't keep messages for subscribers that turn up later. The EURUSD update is missing too, because the subscriber asked only for `stock.`.

## Filter by topic across several subscribers

Every subscriber gets its own copy of each message that matches it. Give the program three subscribers with different prefixes and a few more updates. Replace the `UPDATES` constant with these two constants:

```rust title="src/main.rs"
const SUBSCRIBERS: [(&str, &str); 3] = [
    ("stocks", "stock."),
    ("apple", "stock.AAPL"),
    ("everything", ""),
];

const UPDATES: [(&str, &str); 5] = [
    ("stock.AAPL", "189.25"),
    ("fx.EURUSD", "1.0842"),
    ("stock.MSFT", "411.80"),
    ("stock.AAPL", "189.40"),
    ("fx.USDJPY", "148.12"),
];
```

Then replace `main`:

```rust title="src/main.rs"
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let publisher = ctx.socket(SocketType::Pub)?;
    publisher.bind(PRICES).await?;
    let ready = ctx.socket(SocketType::Pull)?;
    ready.bind(READY).await?;

    let mut subscribers = Vec::new();
    for (name, prefix) in SUBSCRIBERS {
        subscribers.push((name, tokio::spawn(subscriber(ctx.clone(), prefix))));
    }
    wait_for_subscribers(&publisher, &ready, SUBSCRIBERS.len(), "sys.hello").await?;
    println!("{} subscribers are listening", SUBSCRIBERS.len());

    for (topic, price) in UPDATES {
        publish(&publisher, topic, price).await?;
    }
    publish(&publisher, "sys.end", "").await?;

    for (name, handle) in subscribers {
        let seen = handle.await.expect("subscriber panicked")?;
        println!("{name}: {}", seen.join(", "));
    }

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

`subscriber` already takes the prefix as an argument, so nothing else changes. `apple` uses a longer prefix than `stocks`. `everything` subscribes to the empty prefix, which also matches the `sys.` messages that `read_round` skips.

```shell-session
$ cargo run
3 subscribers are listening
stocks: stock.AAPL 189.25, stock.MSFT 411.80, stock.AAPL 189.40
apple: stock.AAPL 189.25, stock.AAPL 189.40
everything: stock.AAPL 189.25, fx.EURUSD 1.0842, stock.MSFT 411.80, stock.AAPL 189.40, fx.USDJPY 148.12
```

Each subscriber receives the updates in the order they were published. The subscriber tasks run concurrently, but `main` joins them in a fixed order and prints afterwards, so these lines always come out the same way.

## Change subscriptions with UNSUBSCRIBE

A SUB socket can add and drop subscriptions while it is connected. `UNSUBSCRIBE` takes the exact prefix you passed to `SUBSCRIBE` and removes it. The change travels to the publisher the same way a subscription does, so it takes effect a moment later on the publisher side.

The readiness check covers that too. The subscriber below drops `stock.MSFT`, then subscribes to a new hello topic, `sys.hello.2`. A subscriber's subscription changes reach the publisher in the order it made them. Once a `sys.hello.2` message gets through, the publisher has also applied the `UNSUBSCRIBE` sent before it. The next round of updates is filtered with the new set.

Change the import to bring in `UNSUBSCRIBE`:

```rust title="src/main.rs"
use rzmq::socket::{SUBSCRIBE, UNSUBSCRIBE};
```

Remove the `SUBSCRIBERS` constant and the `subscriber` function, then add `desk` and replace `main`:

```rust title="src/main.rs"
async fn desk(ctx: Context) -> Result<(Vec<String>, Vec<String>), ZmqError> {
    let prices = ctx.socket(SocketType::Sub)?;
    for topic in ["sys.hello.1", "sys.end", "stock.AAPL", "stock.MSFT"] {
        prices.set_option(SUBSCRIBE, topic).await?;
    }
    prices.connect(PRICES).await?;

    let ready = ctx.socket(SocketType::Push)?;
    ready.connect(READY).await?;

    confirm(&prices, &ready, "sys.hello.1").await?;
    let first = read_round(&prices).await?;

    prices.set_option(UNSUBSCRIBE, "stock.MSFT").await?;
    prices.set_option(UNSUBSCRIBE, "sys.hello.1").await?;
    prices.set_option(SUBSCRIBE, "sys.hello.2").await?;

    confirm(&prices, &ready, "sys.hello.2").await?;
    let second = read_round(&prices).await?;
    Ok((first, second))
}

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

    let publisher = ctx.socket(SocketType::Pub)?;
    publisher.bind(PRICES).await?;
    let ready = ctx.socket(SocketType::Pull)?;
    ready.bind(READY).await?;

    let desk = tokio::spawn(desk(ctx.clone()));

    for round in ["sys.hello.1", "sys.hello.2"] {
        wait_for_subscribers(&publisher, &ready, 1, round).await?;
        for (topic, price) in UPDATES {
            publish(&publisher, topic, price).await?;
        }
        publish(&publisher, "sys.end", "").await?;
    }

    let (first, second) = desk.await.expect("desk panicked")?;
    println!("round 1: {}", first.join(", "));
    println!("round 2: {}", second.join(", "));

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

The desk subscribes to exact hello topics this time, `sys.hello.1` and later `sys.hello.2`. A broad `sys.` prefix would let round-two hellos through before the unsubscribe is in place, which would defeat the check. The publisher sends the same five updates in both rounds.

```shell-session
$ cargo run
round 1: stock.AAPL 189.25, stock.MSFT 411.80, stock.AAPL 189.40
round 2: stock.AAPL 189.25, stock.AAPL 189.40
```

MSFT drops out of the second round while AAPL keeps coming.

## Let a slow subscriber fall behind with SNDHWM

A publisher can't wait for its slowest subscriber without holding up everyone else. The `SNDHWM` option (send high-water mark) caps how many messages queue up for each connected peer. When a peer's queue reaches the mark, the socket reports a `ConnectionCongested` monitor event and skips that peer until its queue drains below the mark, when `ConnectionUncongested` follows. For a PUB socket that means the slow subscriber misses updates while the publisher and the other subscribers carry on.

This step sets `SNDHWM` to 1000 before `bind` and publishes 100,000 updates to two subscribers. The fast one reads as the updates arrive. The slow one confirms it is listening and then reads nothing until the publisher has finished. Both subscribers set `RCVTIMEO` to 500 ms. When nothing arrives for half a second, `recv_multipart` returns `ZmqError::Timeout` and the subscriber stops counting, so one that misses `sys.end` still finishes.

Replace the imports at the top of the file:

```rust title="src/main.rs"
use rzmq::socket::{RCVTIMEO, SNDHWM, SUBSCRIBE, SocketEvent};
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use std::time::Duration;
use tokio::sync::oneshot;
```

Replace the `UPDATES` constant:

```rust title="src/main.rs"
const UPDATE_COUNT: u32 = 100_000;
```

Replace `read_round` and `desk` with these two functions:

```rust title="src/main.rs"
async fn listen(ctx: &Context) -> Result<Socket, ZmqError> {
    let prices = ctx.socket(SocketType::Sub)?;
    prices.set_option(SUBSCRIBE, "").await?;
    prices.set_option(RCVTIMEO, 500i32).await?;
    prices.connect(PRICES).await?;

    let ready = ctx.socket(SocketType::Push)?;
    ready.connect(READY).await?;

    confirm(&prices, &ready, "sys.hello").await?;
    Ok(prices)
}

async fn count_updates(prices: &Socket) -> Result<u32, ZmqError> {
    let mut count = 0;
    loop {
        match prices.recv_multipart().await {
            Ok(frames) => match text(&frames[0]).as_str() {
                "sys.end" => return Ok(count),
                "sys.hello" => {}
                _ => count += 1,
            },
            Err(ZmqError::Timeout) => return Ok(count),
            Err(e) => return Err(e),
        }
    }
}
```

Then replace `main`:

```rust title="src/main.rs"
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let publisher = ctx.socket(SocketType::Pub)?;
    publisher.set_option(SNDHWM, 1000i32).await?;
    let events = publisher.monitor_default().await?;
    publisher.bind(PRICES).await?;
    let ready = ctx.socket(SocketType::Pull)?;
    ready.bind(READY).await?;

    let congestion = tokio::spawn(async move {
        while let Ok(event) = events.recv().await {
            if let SocketEvent::ConnectionCongested { .. } = event {
                return true;
            }
        }
        false
    });

    let fast_ctx = ctx.clone();
    let fast = tokio::spawn(async move {
        let prices = listen(&fast_ctx).await?;
        count_updates(&prices).await
    });

    let (wake, woken) = oneshot::channel::<()>();
    let slow_ctx = ctx.clone();
    let slow = tokio::spawn(async move {
        let prices = listen(&slow_ctx).await?;
        woken.await.expect("main dropped the wake-up sender");
        count_updates(&prices).await
    });

    wait_for_subscribers(&publisher, &ready, 2, "sys.hello").await?;
    for n in 0..UPDATE_COUNT {
        publish(&publisher, "stock.AAPL", &n.to_string()).await?;
        if n % 100 == 99 {
            tokio::task::yield_now().await;
        }
    }
    publish(&publisher, "sys.end", "").await?;
    println!("published {UPDATE_COUNT} updates");

    let fast_count = fast.await.expect("fast subscriber panicked")?;
    println!("fast subscriber got {fast_count}");

    wake.send(()).expect("slow subscriber is gone");
    let slow_count = slow.await.expect("slow subscriber panicked")?;
    println!("slow subscriber got {slow_count}");

    if congestion.await.expect("monitor task panicked") {
        println!("the publisher reported a congested subscriber");
    }

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

The monitor task stops at the first `ConnectionCongested` event. The oneshot channel keeps the slow subscriber from reading until `main` wakes it. The publish loop yields every 100 sends because the `current_thread` runtime runs the fast subscriber's task only when `main` gives up the thread. A send to a full peer is skipped without waiting, so an unpaced loop never yields, the fast subscriber's queue fills up too and it misses updates as well.

```shell-session
$ cargo run
published 100000 updates
fast subscriber got 100000
slow subscriber got 31700
the publisher reported a congested subscriber
```

The slow subscriber's count can differ on your machine. It depends on how much the operating system's TCP buffers take in before the high-water mark is reached. What stays fixed is that the publisher finishes, the fast subscriber gets every update and the slow one gets fewer. A subscriber that needs every update has to keep up with the feed or fetch what it missed from somewhere else.

## Shut down cleanly

Every task in these programs knows when it is done, so shutdown is a matter of waiting in the right order. This is the end of `main` from the topic filtering step:

```rust title="src/main.rs"
    publish(&publisher, "sys.end", "").await?;

    for (name, handle) in subscribers {
        let seen = handle.await.expect("subscriber panicked")?;
        println!("{name}: {}", seen.join(", "));
    }

    ctx.term().await?;
```

The publisher sends `sys.end` after its last update. Messages from one publisher reach a subscriber in order, so each subscriber sees `sys.end` after everything it subscribed to and returns. `main` waits for every subscriber and then calls `Context::term()`, which closes the PUB, SUB, PUSH and PULL sockets made from the context and waits for their background tasks. Errors go up with `?` instead of being ignored.

## Where to go next

The usage guide covers the rest of the socket options, monitoring events, security (PLAIN, CURVE and Noise_XX) and io_uring on Linux in the [rzmq usage guide](https://github.com/excsn/rzmq/blob/main/core/README.USAGE.md) on GitHub.

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