bind

ZmqEndpoint::bind makes this process listen: one subscription reads the bound socket, and port zero lets the system pick.

A bound endpoint is one listening socket, and it belongs to the subscription that bound it. A second subscription on the same broker fails startup with an error that names both subscriptions. So two kinds of work take two brokers, each on an endpoint of its own. When the subscription closes, the endpoint is free for the next one.

The subscription binds its socket as it opens, and a port another process holds fails startup, as Startup describes. With port zero, as in tcp://127.0.0.1:0, the system picks the port when the subscription binds. A publisher of the same service dials the port the system picked, so a service that holds both ends runs with no fixed port.

A broker that binds an endpoint and has no subscription on it listens through its publisher from the moment the service starts, and the peers dial it before its first publish. When a subscription of the same broker binds the endpoint later, the publisher gives the endpoint up first, on a fixed port and on port zero alike, and its next publish dials that subscription. So a handler that publishes into a queue whose subscription opens after its own reaches that subscription once it is open. On a bound endpoint that a subscription of this service reads, a ZmqQueue publisher sends only under the subscription's own name.

Pipelines with PUSH/PULL decides which side of a pipeline binds.

Example

A worker listens for jobs on one endpoint and for reports on another. Each broker is labeled, so a test addresses it by name.

Macros
Manual

examples/zmq_broker_bind.rs
/// The service on the two addresses it listens on, one per subscription.
fn service(jobs: &str, reports: &str) -> RustStream {
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled("jobs", ZmqQueue::new(ZmqEndpoint::bind(jobs)), |b| {
            b.include(work);
        })
        .with_broker_labeled("reports", ZmqQueue::new(ZmqEndpoint::bind(reports)), |b| {
            b.include(file_report);
        })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service("tcp://0.0.0.0:5555", "tcp://0.0.0.0:5556")
}

examples/manual/zmq_broker_bind.rs
/// The service on the two addresses it listens on, one per subscription.
fn service(jobs: &str, reports: &str) -> RustStream {
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled("jobs", ZmqQueue::new(ZmqEndpoint::bind(jobs)), |b| {
            b.include(subscriber("jobs", Work).build());
        })
        .with_broker_labeled("reports", ZmqQueue::new(ZmqEndpoint::bind(reports)), |b| {
            b.include(subscriber("reports", FileReport).build());
        })
}

fn app() -> impl App<State = ()> {
    service("tcp://0.0.0.0:5555", "tcp://0.0.0.0:5556")
}

Example​

Example