Startup

connect checks the endpoint without I/O; each subscription attaches its socket as it opens, and a refused bind fails the start naming it.

When the runtime connects a broker, the connect checks the endpoint's scheme and opens nothing. An address that is neither tcp:// nor ipc:// fails startup with ZmqError::Invalid, naming the address.

Each subscription opens its own socket as the runtime opens the subscription. A subscription on a bound endpoint binds it, and a port another process holds fails startup with ZmqError::Endpoint, naming the address. A subscription that dials waits for its peer to listen. The ZeroMQ client retries the connection for up to 30 seconds, then the start fails with ZmqError::Endpoint, naming the address. A publisher on the bind side attaches its socket at startup, so it listens before its first publish; a publisher that dials attaches on its first publish, the same way.

The sockets attach, and their drivers run, on the runtime the broker connected on. So a handler that runs on threads of its own, mounted with threads(n), publishes through the same sockets as every other handler.

Running in production covers the order in which services start.

Example

The worker binds the address service takes, and opens it when the runtime starts the subscription. A test starts the same service on an address that fails, to see the error the service would stop with.

Macros
Manual

examples/zmq_broker_startup.rs
/// The worker on the address `jobs` names; nothing is opened until the runtime starts it.
fn service(jobs: &str) -> RustStream {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind(jobs)),
        |b| {
            b.include(work);
        },
    )
}

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

examples/manual/zmq_broker_startup.rs
/// The worker on the address `jobs` names; nothing is opened until the runtime starts it.
fn service(jobs: &str) -> RustStream {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind(jobs)),
        |b| {
            b.include(subscriber("jobs", Work).build());
        },
    )
}

fn app() -> impl App {
    service("tcp://0.0.0.0:5555")
}

Example​

Example