Errors

PubSubError is the one error type of the crate: connect, admin, receive, publish, not connected, invalid descriptor and a delay past the lease.

Each variant names the step that failed. The client library's error behind it is its source, so the library's types stay out of the crate's API.

Variant	When
`Connect`	a client could not be built
`Admin`	a topic or subscription call at startup failed, such as a create without permission
`Receive`	a streaming pull failed for good, such as a subscription that does not exist
`Publish`	the service refused a message, or the crate refused one it cannot send, such as an attribute that is not UTF-8
`NotConnected`	a publisher was used after the broker shut down
`InvalidDescriptor`	a subscription the crate cannot open: an empty name, half a dead-letter declaration, a cap outside 5 to 100
`DelayBeyondLease`	a `retry_after` asked for more than the subscription's `max_lease`

A service meets the error in three places. A failed startup carries it as the source of the runtime's error, and the service exits before it reads anything. A publish from a handler returns it inside the core's PublishError, and the handler decides how the delivery settles. A settlement the subscription refuses, such as a delay past the lease, reaches the runtime as the source of an AckError. The enum is #[non_exhaustive], so a match on it ends with a wildcard arm.

Example

A shipment whose note exceeds the 1024 bytes an attribute holds is refused; the handler logs the cause and drops the order.

Macros
Manual

examples/pubsub_broker_errors.rs
/// Ships an order with its note as an attribute of the shipment.
#[subscriber("orders-workers")]
async fn ship(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("note", order.note.clone());
    let shipment = Shipment { order_id: order.id };
    if let Err(err) = out.message(&shipment).with_headers(headers).publish().await {
        // The error names the topic and the service's reason. The same shipment would be
        // refused again, so the order is dropped rather than retried.
        tracing::warn!(subscription = "orders-workers", order.id, error = %err, "shipment refused");
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

examples/manual/pubsub_broker_errors.rs
/// Ships an order with its note as an attribute of the shipment.
struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert("note", order.note.clone());
        let shipment = Shipment { order_id: order.id };
        let sent = outs
            .get(DefaultSlot)
            .message(&shipment)
            .with_headers(headers)
            .publish()
            .await;
        if let Err(err) = sent {
            // The error names the topic and the service's reason. The same shipment would be
            // refused again, so the order is dropped rather than retried.
            tracing::warn!(subscription = "orders-workers", order.id, error = %err, "shipment refused");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

Example​

Example