Cancellation

A queue publish is cancel-safe; a fan-out publish, a reply and a request are not, so give up through a timeout, not a select! arm.

A publish waits while the peer applies back-pressure, so a caller may drop it before it completes. On ZmqQueue that is safe. A publish dropped before the socket took its message sends nothing. One dropped in the middle of the send leaves its message with the publisher, and the next publish through it completes that send before its own.

A fan-out publish and a reply are not cancel-safe. Dropping one can leave the message half handed to the socket. On a reply, the requester then waits out its timeout. Dropping a request closes the DEALER socket it went out on: an answer already on its way is lost, and the request itself may have reached the responder.

So a handler gives up on a request through its timeout, which ends the wait cleanly and returns ZmqError::RequestTimeout. A fan-out publish or a reply that must not hold up other work runs in a task of its own rather than inside a select! arm.

Request and reply gives up on a request through its timeout and asks again later.

Example

The handler asks the pricing service for a quote and waits at most two seconds. With no quote in time it settles the order as a retry in thirty seconds.

Macros
Manual

examples/zmq_publishing_cancellation.rs
#[subscriber("orders")]
async fn price(order: &Order, Out(pricing): Out<impl RequestReply>) -> HandlerOutcome {
    let Ok(ask) = serde_json::to_vec(&Ask { id: order.id }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("quotes", ask.as_slice());
    // The timeout is how the handler gives up: dropping a pending request instead would close
    // its DEALER and lose an answer already on its way.
    match pricing.request(request, QUOTE_TIMEOUT).await {
        Ok(answer) => {
            let Ok(quote) = serde_json::from_slice::<Quote>(answer.payload()) else {
                return HandlerOutcome::drop();
            };
            tracing::info!(order.id, quote.price, "priced");
            HandlerOutcome::ack()
        }
        Err(err) => {
            tracing::warn!(subscription = "orders", order.id, error = %err, "no quote");
            HandlerOutcome::retry_after(Duration::from_secs(30))
        }
    }
}

examples/manual/zmq_publishing_cancellation.rs
struct Price;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Price
where
    Entry: OutEntry<DefaultSlot, Wire: RequestReply>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let ask = serde_json::to_vec(&Ask { id: order.id }).map_err(|_| HandlerOutcome::drop())?;
        let request = OutgoingMessage::new("quotes", ask.as_slice());
        // The timeout is how the handler gives up: dropping a pending request instead would close
        // its DEALER and lose an answer already on its way.
        let answer = outs
            .get(DefaultSlot)
            .request(request, QUOTE_TIMEOUT)
            .await
            .map_err(|err| {
                tracing::warn!(subscription = "orders", order.id, error = %err, "no quote");
                HandlerOutcome::retry_after(Duration::from_secs(30))
            })?;
        let quote = serde_json::from_slice::<Quote>(answer.payload())
            .map_err(|_| HandlerOutcome::drop())?;
        tracing::info!(order.id, quote.price, "priced");
        Ok(())
    }
}

Example​

Example