ZmqEndpoint

ZmqEndpoint is an address on tcp:// or ipc:// plus the side this process takes; another scheme is refused at connect.

Two constructors fix the side. ZmqEndpoint::bind makes this process listen, and ZmqEndpoint::connect makes it dial. The side is part of the type, and the broker built on the endpoint carries it, as EndpointRole describes.

tcp://host:port reaches another host. ipc:///path is a socket file on the same host, with no network stack in between. When the broker connects, an address with any other scheme fails startup with ZmqError::Invalid, naming the address, before any socket opens. address returns the address as written.

The endpoint and its side are the whole configuration of a broker. The ZeroMQ implementation carries no CURVE or ZAP layer, so traffic is neither authenticated nor encrypted: a service runs on a trusted network or inside a tunnel.

Running in production reads the endpoints from the environment.

Example

A worker listens on a socket file for the producers on its host, and pushes its results to a collector it dials over the network.

Macros
Manual

examples/zmq_broker_endpoint.rs
#[ruststream::app]
fn app() -> impl App {
    // This process dials the collector, which listens on another host.
    let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
    let results = collector.bind(Publish);
    // This process listens, on a socket file the producers of this host dial.
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")), |b| {
            b.include(work).out_reply(results);
        })
        .register_broker(collector)
}

examples/manual/zmq_broker_endpoint.rs
fn app() -> impl App {
    // This process dials the collector, which listens on another host.
    let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
    let results = collector.bind(Publish);
    // This process listens, on a socket file the producers of this host dial.
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")), |b| {
            b.include(subscriber("jobs", Work).reply().build())
                .out_reply(results);
        })
        .register_broker(collector)
}

Example​

Example