nng-rs / Rust pattern guide

pubsub communication topology

PUB / SUB — Topics and Broadcast

PUB / SUB broadcasts messages, while subscribers choose the topics they receive.

nng-rs is an independently maintained Rust binding for NNG. Its maintainers and release cycle are separate from the NNG core project.

This guide is based on the project’s default-branch example, with the required empty-topic subscription enabled through nng-rs’s optional FFI module. See the upstream example for its current history.

pubsub.rs

//! Pub/Sub (topics & broadcast) example.
//!
//! This pattern is used to allow a single broadcaster to publish messages to many subscribers,
//! which may choose to limit which messages they receive.
use nng::{ffi, PipeEvent, Protocol, Socket};
use std::{
    convert::TryInto,
    env, process,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    thread,
    time::Duration,
};

/// Entry point of the application.
fn main() -> Result<(), nng::Error> {
    // Begin by parsing the arguments to determine whether this is the
    // subscriber or the publisher and what URL to connect with.
    let args: Vec<_> = env::args().take(3).collect();

    match &args[..] {
        [_, t, url] if t == "publisher" => publisher(url),
        [_, t, url] if t == "subscriber" => subscriber(url),
        _ => {
            println!("Usage: pubsub publisher|subscriber <url>");
            process::exit(1);
        }
    }
}

/// Run the publisher portion of the program.
fn publisher(url: &str) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Pub0)?;
    let count = Arc::new(AtomicUsize::new(0));
    let count_clone = count.clone();

    s.pipe_notify(move |_, ev| {
        match ev {
            PipeEvent::AddPost => count_clone.fetch_add(1, Ordering::Relaxed),
            PipeEvent::RemovePost => count_clone.fetch_sub(1, Ordering::Relaxed),
            _ => 0,
        };
    })?;

    s.listen(url)?;

    loop {
        // Sleep for a little bit before sending the next message.
        thread::sleep(Duration::from_secs(3));

        // Load the number of subscribers and send the value across
        let data = count.load(Ordering::Relaxed) as u64;
        println!("PUBLISHER: SENDING {}", data);
        s.send(data.to_le_bytes())?;
    }
}

/// Run the subscriber portion of the program.
fn subscriber(url: &str) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Sub0)?;
    s.dial(url)?;

    println!("SUBSCRIBER: SUBSCRIBING TO ALL TOPICS");
    let rv = unsafe { ffi::nng_sub0_socket_subscribe(s.nng_socket(), std::ptr::null(), 0) };
    assert_eq!(rv, 0, "could not subscribe to all topics");

    loop {
        let msg = s.recv()?;
        let subs = u64::from_le_bytes(msg[..].try_into().unwrap());
        println!("SUBSCRIBER: THERE ARE {} SUBSCRIBERS", subs);
    }
}
Run it

# Set up once.
git clone --recurse-submodules https://github.com/nanomsg/nng-rs.git
cd nng-rs

# Run these in separate terminals from the checkout.
cargo run -p nng --features ffi-module --example pubsub -- publisher tcp://127.0.0.1:6004
cargo run -p nng --features ffi-module --example pubsub -- subscriber tcp://127.0.0.1:6004