Подтверждение и гарантии доставки

Это руководство урегулирует доставки, ограничивает число попыток, читает очередь в режиме at-most-once, выбирает кредит связи и обрабатывает пакеты.

В AMQP 1.0 получатель урегулирует каждую доставку исходом, и брокер действует по нему. Исход обработчика и есть эта диспозиция.

Пример - склад. Он комплектует отгрузки, записывает каждую комплектацию в журнал и пополняет полки из поставок, которые приходят пакетами.

Три исхода

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Shipment {
    id: u64,
    sku: String,
    quantity: u32,
}

/// The goods the warehouse has on its shelves.
fn in_stock(sku: &str) -> bool {
    matches!(sku, "book" | "pen")
}

#[subscriber(AmqpAddress::queue("shipments").credit(nonzero!(32)))]
async fn pick(shipment: &Shipment) -> HandlerOutcome {
    if shipment.quantity == 0 {
        tracing::warn!(shipment.id, "an empty shipment rejected");
        return HandlerOutcome::drop();
    }
    if !in_stock(&shipment.sku) {
        tracing::info!(
            shipment.id,
            shipment.sku,
            "out of stock, the shipment comes back"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(shipment.id, shipment.sku, "shipment picked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Shipment {
    id: u64,
    sku: String,
    quantity: u32,
}

/// The goods the warehouse has on its shelves.
fn in_stock(sku: &str) -> bool {
    matches!(sku, "book" | "pen")
}

struct Pick;

impl Handle<Shipment> for Pick {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if shipment.quantity == 0 {
            tracing::warn!(shipment.id, "an empty shipment rejected");
            return ready(Err(HandlerOutcome::drop()));
        }
        if !in_stock(&shipment.sku) {
            tracing::info!(
                shipment.id,
                shipment.sku,
                "out of stock, the shipment comes back"
            );
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(shipment.id, shipment.sku, "shipment picked");
        ready(Ok(()))
    }
}

Исход обработчика	Диспозиция	Что делает брокер
HandlerOutcome::ack()	accepted	удаляет сообщение
HandlerOutcome::retry()	modified с delivery-failed	учитывает попытку в delivery-count и доставляет сообщение повторно
HandlerOutcome::drop()	rejected	применяет свою политику недоставленных сообщений

pick принимает отгрузку, которую скомплектовал. Пустую отгрузку скомплектовать невозможно, поэтому pick её отклоняет. Artemis переносит отклонённое сообщение в адрес недоставленных сообщений, по умолчанию DLQ. Отгрузку товаров, которых нет на складе, можно скомплектовать позже, поэтому pick просит доставить её снова. Artemis прекращает доставлять сообщение после max-delivery-attempts попыток, по умолчанию 10, и переносит его в адрес недоставленных сообщений.

Предел попыток

Макросы

src/main.rs

fn service(broker: AmqpBroker) -> RustStream {
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(broker, |b| {
        b.include(pick).max_attempts(nonzero!(5u32));
        b.include(log_pick);
        b.include(restock.batch(nonzero!(20usize)));
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(AmqpBroker::new("amqp://artemis:artemis@localhost:5672"))
}

Вручную

src/main.rs

fn service(broker: AmqpBroker) -> RustStream {
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(broker, |b| {
        let shipments = AmqpAddress::queue("shipments").credit(nonzero!(32));
        b.include(subscriber(shipments, Pick).build())
            .max_attempts(nonzero!(5u32));
        let pick_log = AmqpAddress::queue("pick-log").settle(Settle::AtMostOnce);
        b.include(subscriber(pick_log, LogPick).build());
        let restock = AmqpAddress::queue("restock").batch_wait(Duration::from_millis(50));
        b.include(
            subscriber(restock, Restocking)
                .batch(nonzero!(20usize))
                .build(),
        );
    })
}

fn app() -> impl App {
    service(AmqpBroker::new("amqp://artemis:artemis@localhost:5672"))
}

fn main() -> ExitCode {
    run_main(app)
}

max_attempts ограничивает доставки одной отгрузки пятью, включая первую. Пятая доставка отклоняется, а не повторяется. Число попыток - это delivery-count брокера, если у сообщения есть секция AMQP header. У сообщения, которое публикует этот крейт, её нет. Для такого сообщения фреймворк считает попытки сам: retry публикует копию обратно в shipments с номером повтора в заголовке. Руководство Повторы и недоставленные сообщения добавляет адрес недоставленных сообщений и задержку.

Сбой посреди обработки

По умолчанию подписки работают в режиме at-least-once. Доставка, которую сервис не урегулировал к моменту остановки, возвращается в очередь, и её получает следующий потребитель. Поэтому обработчик может увидеть одно сообщение дважды, и его работа должна это выдерживать.

Не более одного раза

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Picked {
    shipment: u64,
}

#[subscriber(AmqpAddress::queue("pick-log").settle(Settle::AtMostOnce))]
async fn log_pick(picked: &Picked) -> HandlerOutcome {
    tracing::info!(picked.shipment, "pick logged");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Picked {
    shipment: u64,
}

struct LogPick;

impl Handle<Picked> for LogPick {
    fn handle(
        &self,
        picked: &Picked,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(picked.shipment, "pick logged");
        ready(Ok(()))
    }
}

Settle::AtMostOnce урегулирует каждую доставку при получении. Брокер забывает сообщение в тот момент, когда отдаёт его, поэтому сбой до завершения обработчика его теряет. Это подходит журналу комплектаций: пропущенная строка обходится дешевле второго прохода по брокеру. Исходу обработчика урегулировать уже нечего: ack и nack возвращают AckError::Unsupported, и рантайм пишет предупреждение в журнал для каждого.

Кредит

.credit(nonzero!(32)) на shipments - кредит связи: сколько доставок брокер может отправить вперёд, не дожидаясь урегулирования от сервиса. По умолчанию это DEFAULT_CREDIT, 256. Когда урегулирована половина кредита, клиент снова выдаёт полный кредит. Поэтому медленный обработчик сдерживает брокер, а не заполняет память сервиса, и остаток очереди остаётся на брокере для других реплик. Кредит имеет тип NonZeroU32, потому что связь без кредита ничего не получает: credit(nonzero!(0)) не компилируется.

Пакеты

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Restock {
    sku: String,
    quantity: u32,
}

#[subscriber(AmqpAddress::queue("restock").batch_wait(Duration::from_millis(50)))]
async fn restock(items: &[Restock]) -> HandlerOutcome {
    let units: u32 = items.iter().map(|item| item.quantity).sum();
    tracing::info!(lines = items.len(), units, "shelves restocked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Restock {
    sku: String,
    quantity: u32,
}

struct Restocking;

impl Handle<[Restock]> for Restocking {
    fn handle(
        &self,
        items: &[Restock],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let units: u32 = items.iter().map(|item| item.quantity).sum();
        tracing::info!(lines = items.len(), units, "shelves restocked");
        ready(Ok(()))
    }
}

Обработчик, который принимает срез, получает пакет, а .batch(nonzero!(20usize)) в точке монтирования задаёт его наибольший размер. В AMQP 1.0 одна передача содержит одно сообщение, поэтому пакет собирает клиент. Пакет содержит не больше 20 доставок, а меньше - если больше не пришло вовремя. batch_wait - сколько неполный пакет ждёт новых доставок: здесь 50 миллисекунд, по умолчанию 10. Исход restock урегулирует все доставки пакета.

Тестирование исходов

Тесты в файле примера запускают склад внутри процесса. Они проверяют каждый исход, пять доставок отгрузки товара, которого нет на складе, журнал комплектаций и пакеты. Кредит, политика недоставленных сообщений брокера и его предел доставок принадлежат Artemis, поэтому брокер внутри процесса их не моделирует. Руководство Тестирование сервиса на AMQP разбирает обвязку.