Errors

AmqpError is the one error type of the crate: connect, session, attach, publish, receive, request, transaction and a body with no bytes.

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

Variant	When
`Connect`	the URL, the socket, TLS, the login or the AMQP open failed, or the connection failed to close at shutdown
`Session`	the broker refused a session, the publishers' or a subscription's
`Attach`	the broker refused a link: a subscription's receiver, a publisher's sender, a request's reply link
`Detach`	a link failed to close at shutdown
`Publish`	the connection failed while a message was sent
`PublishNotAccepted`	the broker settled a published message as rejected, released or modified, so the message is not on the broker
`Receive`	the connection failed while a subscription received
`UnsupportedBody`	a delivery's body is an AMQP value that is neither binary nor a string; the delivery is rejected, and the runtime logs the error
`RequestTimeout`	a request had no reply within its timeout
`NotConnected`	a publisher, a subscription or a transaction was used before the connect or after the shutdown
`InvalidAddress`	a subscription's address is empty
`Transaction`	a transaction step was called in a state that cannot serve it (`transaction` feature)

A service meets the error in two places. A failed startup carries it as a source of the runtime's error, and the service exits before it reads anything. A publish from a handler returns it as the source of 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

The handler sends each shipment to the warehouse address its order names. A shipment the broker refuses drops the order, since sending it again meets the same answer; any other failure retries it.

Macros
Manual

examples/amqp_broker_errors.rs
#[subscriber(AmqpAddress::queue("orders"))]
async fn ship(order: &Order, Out(warehouses): Out<impl Publisher>) -> HandlerOutcome {
    let shipment = Shipment { order_id: order.id };
    let sent = warehouses
        .message(&shipment)
        .to(order.warehouse.as_str())
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not sent");
            if refused_by_the_broker(&err) {
                HandlerOutcome::drop()
            } else {
                HandlerOutcome::retry()
            }
        }
    }
}

/// Whether the broker settled the message with an outcome other than accepted. Sending it again
/// meets the same answer, while a connection that failed under the publish may be back by the next
/// attempt.
fn refused_by_the_broker(err: &(dyn Error + 'static)) -> bool {
    let mut cause = Some(err);
    while let Some(err) = cause {
        if let Some(AmqpError::PublishNotAccepted { .. }) = err.downcast_ref() {
            return true;
        }
        cause = err.source();
    }
    false
}

examples/manual/amqp_broker_errors.rs
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 shipment = Shipment { order_id: order.id };
        let sent = outs
            .get(DefaultSlot)
            .message(&shipment)
            .to(order.warehouse.as_str())
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not sent");
            if refused_by_the_broker(&err) {
                return Err(HandlerOutcome::drop());
            }
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

/// Whether the broker settled the message with an outcome other than accepted. Sending it again
/// meets the same answer, while a connection that failed under the publish may be back by the next
/// attempt.
fn refused_by_the_broker(err: &(dyn Error + 'static)) -> bool {
    let mut cause = Some(err);
    while let Some(err) = cause {
        if let Some(AmqpError::PublishNotAccepted { .. }) = err.downcast_ref() {
            return true;
        }
        cause = err.source();
    }
    false
}

Example​

Example