Struct ZmqRpc 

pub struct ZmqRpc { /* private fields */ }

The DEALER/ROUTER request-reply pattern.

The endpoint may take either side, and the pattern answers the same on both: a copy of a request has no address of its own, so every responder registration names where its retry copies go.

Examples

A queue worker that asks a model service for a score before it finishes each job:

use std::error::Error;
use std::time::Duration;

use ruststream::OutgoingMessage;
use ruststream::codec::{Codec, JsonCodec};
use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

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

#[derive(Deserialize)]
struct Score {
    value: f64,
}

#[subscriber("jobs")]
async fn rate(
    job: &Job,
    Out(model): Out<impl RequestReply>,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let request = JsonCodec.encode(job)?;
    let reply = model
        .request(OutgoingMessage::new("score", request.as_ref()), Duration::from_secs(5))
        .await?;
    let score: Score = JsonCodec.decode(reply.payload())?;
    tracing::info!(job.id, score = score.value, "scored");
    Ok(())
}

#[ruststream::app]
fn app() -> impl App {
    let model = ZmqRpc::new(ZmqEndpoint::connect("tcp://model:5557")).bindable();
    let ask = model.bind(ZmqRpcPublish);
    RustStream::new(AppInfo::new("scorer", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
            b.include(rate).out(DefaultSlot, ask).build();
        })
        .register_broker(model)
}

Implementations

impl ZmqRpc

pub fn new<Role: EndpointRole>(endpoint: ZmqEndpoint<Role>) -> Self

Records the endpoint, on either side. No I/O.

Examples

A model service that answers each score request at the peer that sent it:

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

#[derive(Deserialize)]
struct Features {
    values: Vec<f64>,
}

#[derive(Serialize, Outgoing)]
struct Score {
    value: f64,
}

#[subscriber("score", reply("reply"))]
async fn score(request: &Features) -> Score {
    Score {
        value: request.values.iter().sum(),
    }
}

// Answers at the request's `reply-to` address; the `ZmqRpcPublisher` documentation shows it
// in full.

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("model", "0.1.0")).with_broker(
        ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            b.include(score)
                .out_reply(Publish)
                .transform(ReplyToRequester)
                .out_retry(Publish)
                .to("score.retry");
        },
    )
}

pub const fn read_ahead(self, read_ahead: NonZeroUsize) -> Self

How many deliveries a subscription reads off its socket ahead of the handler; 1000 unless set, the receive high-water mark ZeroMQ itself gives a socket.

Past this bound the subscription stops reading, the socket’s buffers fill, and the sender waits: a handler slower than the wire slows the sender down rather than growing this process’s memory.

Examples

A responder whose answers take a database round trip holds 64 requests ahead, and the requesters wait beyond that:

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

#[derive(Deserialize)]
struct Lookup {
    customer: u64,
}

#[derive(Serialize, Outgoing)]
struct Customer {
    name: String,
}

#[subscriber("customers", reply("reply"))]
async fn lookup(request: &Lookup) -> Customer {
    Customer {
        name: format!("customer {}", request.customer),
    }
}

// Answers at the request's `reply-to` address; the `ZmqRpcPublisher` documentation shows it
// in full.

#[ruststream::app]
fn app() -> impl App {
    let requests = ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557"))
        .read_ahead(nonzero!(64_usize));
    RustStream::new(AppInfo::new("customers", "0.1.0")).with_broker(requests, |b| {
        b.include(lookup)
            .out_reply(Publish)
            .transform(ReplyToRequester)
            .out_retry(Publish)
            .to("customers.retry");
    })
}

pub fn publisher(&self) -> ZmqRpcPublisher

A publisher sharing this pattern’s state; buildable before connect.

Trait Implementations

impl Broker for ZmqRpc

type Error = ZmqError

The error type returned by broker-level operations.

type Connected = ConnectedZmqRpc

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl Clone for ZmqRpc

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ZmqRpc

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DescribeServer for ZmqRpc

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl InProcess for ZmqRpc

async fn connect_in_process(self) -> Result<Self::Connected, Self::Error>

Connects to the in-process transport, consuming the configuration.