ruststream-zeromq 0.7.1

ruststream-zeromq connects a RustStream service to a ZeroMQ topology over the pure-Rust zeromq implementation, on TCP and IPC. Peers talk directly, so a Rust service joins sockets a Python worker or a C++ daemon already speaks. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Three socket patterns: ZmqQueue (PUSH/PULL, competing consumers), ZmqFanout (PUB/SUB, prefix filtering) and ZmqRpc (DEALER/ROUTER, request and reply).
• An explicit role: an endpoint either binds or connects, which is a deployment decision.
• A stable wire contract that a foreign peer composes by hand.
• Batches assembled on the client for the one-way patterns.
• Retry caps and dead letters applied by the framework.
• AsyncAPI with the transport, role and socket pair, behind the asyncapi feature.
• Tests without sockets: the service's own app runs with this crate's brokers in process.

Delivery is at most once and nothing is stored: acknowledgement reports AckError::Unsupported, and a subscriber that connects late misses what was sent before it arrived. Traffic travels in the clear, so a service runs on a trusted network or inside a tunnel.

The wire contract

frame 0: name      UTF-8; also the subscription prefix for the fan-out pattern
frame 1: headers   UTF-8 "name: value" lines separated by \n; may be empty
frame 2: payload   encoded by the framework's codec

A Python peer sends socket.send_multipart([b"orders", b"content-type: application/json", payload]). A two-frame message reads as one without headers.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-zeromq = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-zeromq = { version = "0.7", features = ["testing"] }

Write a service

use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, Serialize)]
struct Job {
    id: u64,
}

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Done {
    id: u64,
}

#[subscriber("jobs", reply("results"))]
async fn handle(job: &Job) -> Done {
    Done { id: job.id }
}

#[ruststream::app]
fn app() -> impl App {
    // The results travel on a queue of their own, which a sink binds.
    let results = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let to_results = results.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
            b.include(handle).out_reply(to_results);
        })
        .with_broker(results, |_b| {})
}

#[ruststream::app] generates main, so the binary understands run and asyncapi gen. Each pattern has its own prelude (queue, fanout, rpc), so switching pattern changes the import.

Test it

TestApp runs the service's own app with this crate's brokers in process, with no sockets; TestApp::start_live(app()) runs the same body over loopback sockets. Enable the testing feature in [dev-dependencies].

use ruststream::testing::TestApp;
use ruststream_zeromq::Connect;

let tb = TestApp::start(app()).await?;

// A foreign peer's push; the injection returns once the handler has settled.
tb.broker::<ZmqQueue>()
    .message(&Job { id: 1 })
    .to("jobs")
    .publish()
    .await?;

tb.broker::<ZmqQueue<Connect>>()
    .published::<Done>("results")
    .assert_called_once()
    .with(&Done { id: 1 });

Documentation

• Getting started - installation and a first worker fed by a Python peer.
• Guides - pipelines, broadcast, request and reply, retries, peers in other languages, testing and production.
• Components - one page per part of the crate: the brokers and endpoints, subscriptions, acknowledgement, publishing, request-reply, the wire format and testing.
• Architecture - how the sockets work and where the crate sits between them and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs.
• The framework: https://powersemmi.github.io/ruststream/latest/

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.