Errors

PulsarError is the crate's one error type: connect, subscribe, receive, publish, an invalid descriptor and a closed connection.

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

Variant	When
`Connect`	the client could not parse the service URL or connect with it
`Subscribe`	the server refused a consumer, naming the topic or the pattern
`Receive`	an open consumer's stream failed or ended
`Publish`	a producer could not open, the server refused a message, the payload is over `max_message_size`, or a header value is not UTF-8 text
`NotConnected`	a publisher or a seek was used after the broker shut down
`Invalid`	a descriptor or a destination is not valid: an empty name, a malformed topic, a pattern that is no regular expression, a bare name with no default subscription

A service meets the error in two 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. The enum is #[non_exhaustive], so a match on it ends with a wildcard arm.

Example

A shipment the broker refuses is logged with its cause, and the order is settled as a retry. After three deliveries the order moves to its dead-letter topic.

Macros
Manual

examples/pulsar_broker_errors.rs
#[subscriber(PulsarSubscription::new("orders", "shipping"))]
async fn ship(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> HandlerOutcome {
    let shipment = Shipment {
        order_id: order.id,
        labels: (1..=order.parcels)
            .map(|parcel| format!("{}-{parcel:06}", order.id))
            .collect(),
    };
    if let Err(err) = shipments.message(&shipment).publish().await {
        // The error names the topic and the reason.
        tracing::warn!(subscription = "shipping", order.id, error = %err, "shipment not published");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/pulsar_broker_errors.rs
struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Shipments, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipment = Shipment {
            order_id: order.id,
            labels: (1..=order.parcels)
                .map(|parcel| format!("{}-{parcel:06}", order.id))
                .collect(),
        };
        if let Err(err) = outs.get(Shipments).message(&shipment).publish().await {
            // The error names the topic and the reason.
            tracing::warn!(subscription = "shipping", order.id, error = %err, "shipment not published");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Example​

Example