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:

rzmq = "0"

Set up the project#

Start a new binary crate.

cargo new pricefeed
cd pricefeed

Add rzmq and Tokio to Cargo.toml:

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:

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:

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

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:

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.

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

src/main.rs

use rzmq::socket::{SUBSCRIBE, UNSUBSCRIBE};

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

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.

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

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:

src/main.rs

const UPDATE_COUNT: u32 = 100_000;

Replace read_round and desk with these two functions:

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:

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.

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

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 on GitHub.

The full API reference is on docs.rs.