Publishing

Each pattern's Publish sends the three frames and nothing else; a send waits up to five seconds for a peer.

Each pattern ships one publish policy: ZmqQueuePublish, ZmqFanoutPublish and ZmqRpcPublish. The pattern's prelude offers it as Publish, and the crate prelude offers all three under their own names. A mount site names it wherever a registration publishes: .out_reply(Publish) for a reply, .out(Slot, Publish).build() for a slot, .out_retry(Publish) for retry copies. The policy holds no connection. At startup the runtime pairs it with the connected broker into a live publisher, such as ZmqQueuePublisher. To publish through another broker of the service, the mount site binds the policy to it with bindable() and .bind(Publish). Pipelines with PUSH/PULL publishes a worker's results to a sink this way.

A publish sends the message as the three frames: its name, its headers and the payload the codec wrote. The payload frame is the buffer the codec wrote, so a payload of any size is sent without a copy. A ZMTP send takes the frames and nothing else, so every publisher declares Options = (), and the publish builder carries the core's steps alone. A handler body therefore imports ruststream::prelude::* alone and bounds its publisher with the capability it needs.

A publisher on the bind side attaches its socket when the service starts, so a peer that dials it connects before the first message. A publisher that dials attaches on its first publish and waits for a peer. When no peer has completed the handshake within five seconds, the publish fails with ZmqError::Send, "no connected peer". The window is a constant of the crate.

Who receives the message is the pattern's rule. On ZmqQueue one peer takes it, in turn. On ZmqFanout every subscriber whose filter is a prefix of the name takes a copy, and a message no subscriber matches is dropped without an error. On ZmqRpc the publisher sends replies only, and a request goes through request. A publish is one send of one message, and a queue publish is safe to cancel, as Cancellation describes.

Example

A worker announces each shipped order on a fan-out it binds itself. A peer that subscribes to orders.shipped, or to the prefix orders, receives every announcement.

Macros
Manual

examples/zmq_publishing_publish.rs
/// The slot shipments leave through, and what it may publish.
#[derive(OutSlot)]
#[publishes(Shipped)]
struct Events;

#[subscriber("orders")]
async fn ship(order: &Order, Out(events): Out<impl Publisher, Events>) -> HandlerOutcome {
    tracing::info!(order.id, "shipping");
    // A send waits up to five seconds for a subscriber to attach, then fails.
    if let Err(err) = events.message(&Shipped { id: order.id }).publish().await {
        tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not announced");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/zmq_publishing_publish.rs
#[ruststream::app]
fn app() -> impl App {
    // No subscription of this service reads the fan-out, so its publisher binds the endpoint.
    let events = ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("shipping", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(ship).out(Events, announce).build();
            },
        )
        .register_broker(events)
}

examples/manual/zmq_publishing_publish.rs
/// The slot shipments leave through.
struct Events;

impl OutSlot for Events {
    const NAME: &'static str = "Events";
    // `Shipped` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;
}

impl PublishedThrough<Events> for Shipped {}

struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Events, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(order.id, "shipping");
        // A send waits up to five seconds for a subscriber to attach, then fails.
        let shipped = Shipped { id: order.id };
        if let Err(err) = outs.get(Events).message(&shipped).publish().await {
            tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not announced");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

examples/manual/zmq_publishing_publish.rs
fn app() -> impl App {
    // No subscription of this service reads the fan-out, so its publisher binds the endpoint.
    let events = ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("shipping", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(subscriber("orders", Ship).build())
                    .out(Events, announce)
                    .build();
            },
        )
        .register_broker(events)
}

Example​

Example