Надёжная публикация

Это руководство выбирает, насколько сервис уверен, что RabbitMQ принял его сообщения, и задаёт свойства отдельного сообщения.

Гарантию задаёт политика публикации в точке монтирования, а не флаг каждого вызова:

Политика	Имя в прелюдии	Публикация завершается, когда
LapinPublish	Publish	кадр записан в соединение
ConfirmsPublish	TransactionalPublish	RabbitMQ подтвердил сообщение
ServerTxPublish	ServerTxPublish	транзакция канала зафиксирована

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

Транзакции с подтверждениями

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    parcels: Vec<String>,
    expedited: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
    parcel: String,
    expedited: bool,
}

/// Publishes one shipment per parcel: every one of them, or none.
async fn ship_all(
    shipments: &impl TransactionalPublisher,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let scope = shipments.begin().await?;
    for parcel in &order.parcels {
        if parcel.is_empty() {
            scope.abort().await?;
            return Err("a parcel without a description".into());
        }
        let shipment = Shipment {
            order_id: order.id,
            parcel: parcel.clone(),
            expedited: order.expedited,
        };
        scope.message(&shipment).publish().await?;
    }
    // Resolves once RabbitMQ confirmed every shipment of the order.
    scope.commit().await?;
    Ok(())
}

#[subscriber("orders")]
async fn pack_order(
    order: &Order,
    Out(shipments): Out<impl TransactionalPublisher>,
) -> HandlerOutcome {
    match ship_all(shipments, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "order not shipped");
            HandlerOutcome::drop()
        }
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    parcels: Vec<String>,
    expedited: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
    parcel: String,
    expedited: bool,
}

/// Publishes one shipment per parcel: every one of them, or none.
async fn ship_all(
    shipments: &impl TransactionalPublisher,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let scope = shipments.begin().await?;
    for parcel in &order.parcels {
        if parcel.is_empty() {
            scope.abort().await?;
            return Err("a parcel without a description".into());
        }
        let shipment = Shipment {
            order_id: order.id,
            parcel: parcel.clone(),
            expedited: order.expedited,
        };
        scope.message(&shipment).publish().await?;
    }
    // Resolves once RabbitMQ confirmed every shipment of the order.
    scope.commit().await?;
    Ok(())
}

struct PackOrder;

impl<Shipments> Handle<Order, (), Outs<(Shipments,)>> for PackOrder
where
    Shipments: OutEntry<DefaultSlot, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Shipments,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        ship_all(outs.get(DefaultSlot), order).await.map_err(|err| {
            tracing::warn!(order.id, %err, "order not shipped");
            HandlerOutcome::drop()
        })
    }
}

Слот pack_order ограничен типом Out<impl TransactionalPublisher>, поэтому обработчик монтируется только на политику с транзакциями. begin открывает область транзакции, и каждое отправление, опубликованное в ней, ждёт в сервисе. commit публикует их и завершается, когда RabbitMQ подтвердил каждое. Посылка без описания прерывает область, и ни одно отправление этого заказа не уходит. Если фиксация не удалась, она возвращает ошибку, и заказ можно доставить и упаковать заново.

RabbitMQ подтверждает сообщение, когда берёт за него ответственность: записывает на диск постоянное сообщение в устойчивой очереди или получает согласие большинства кворумной очереди. Когда подтверждается каждое сообщение, описывает руководство по подтверждениям RabbitMQ.

Подтверждение не значит, что сообщение попало в очередь. Сообщение, которое не подходит ни под одну привязку, RabbitMQ подтверждает и отбрасывает, потому что крейт публикует без флага mandatory. Поэтому подтверждённая публикация должна идти в обменник с существующими привязками или в существующую очередь. Объявите их вместе с сервисом (Топология) или проверьте там, где ими владеет эксплуатация.

Транзакции канала

ServerTxPublish, который даёт Publish::default().server_tx(), вместо этого использует транзакции канала AMQP. Сообщения транзакции становятся видны вместе при commit, а это синхронный обмен с сервером. Подтверждения быстрее и подходят для большинства случаев. Транзакция канала нужна, когда потребители не должны увидеть половину группы, например только одну из двух проводок перевода.

Обработчик, ограниченный Out<impl TransactionalPublisher>, монтируется на любую из двух политик без изменений. ConfirmsTransaction, открытая через owned_transaction, владеет своим буфером, а не заимствует издателя: можно открыть несколько таких транзакций сразу, и каждая фиксируется отдельно. Транзакции канала дают только заимствующую область, потому что транзакция - это состояние самого канала.

Свойства сообщения

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "pickups")]
struct Pickup {
    order_id: u64,
    parcel: String,
}

/// The slot pickups leave through.
#[derive(OutSlot)]
#[publishes(Pickup)]
struct Pickups;

/// An expedited pickup jumps the couriers' queue and is worthless after an hour.
#[subscriber(RabbitQueue::new("shipments"))]
async fn book_courier(
    shipment: &Shipment,
    Out(pickups): Out<impl Publisher<Options = LapinPublishOptions>, Pickups>,
) -> HandlerOutcome {
    let pickup = Pickup {
        order_id: shipment.order_id,
        parcel: shipment.parcel.clone(),
    };
    let sent = if shipment.expedited {
        pickups
            .message(&pickup)
            .priority(9)
            .expiration(Duration::from_hours(1))
            .publish()
            .await
    } else {
        pickups.message(&pickup).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "pickups")]
struct Pickup {
    order_id: u64,
    parcel: String,
}

/// The slot pickups leave through.
#[derive(OutSlot)]
#[publishes(Pickup)]
struct Pickups;

/// An expedited pickup jumps the couriers' queue and is worthless after an hour.
struct BookCourier;

impl<Couriers> Handle<Shipment, (), Outs<(Couriers,)>> for BookCourier
where
    Couriers: OutEntry<Pickups, Wire: Publisher<Options = LapinPublishOptions>>,
{
    async fn handle(
        &self,
        shipment: &Shipment,
        outs: &Outs<(Couriers,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let pickups = outs.get(Pickups);
        let pickup = Pickup {
            order_id: shipment.order_id,
            parcel: shipment.parcel.clone(),
        };
        let sent = if shipment.expedited {
            pickups
                .message(&pickup)
                .priority(9)
                .expiration(Duration::from_hours(1))
                .publish()
                .await
        } else {
            pickups.message(&pickup).publish().await
        };
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

AMQP даёт каждому сообщению три собственных свойства: приоритет, срок жизни и режим доставки. Политика задаёт значения для каждого сообщения, которое через неё проходит, здесь приоритет 3. Шаги билдера публикации задают их для одного сообщения: срочный вызов курьера уходит с приоритетом 9, и если его не прочитали за час, срок его жизни истекает. Слот ограничен типом LapinPublishOptions, поэтому обработчик не называет тип издателя. Шаги берутся из прелюдии крейта.

Ни одно из трёх свойств не передаётся заголовком: каждое - свойство AMQP, откуда его и читает RabbitMQ. Доставка возвращает их в виде заголовков PRIORITY_HEADER и EXPIRATION_HEADER.

Приоритет упорядочивает доставки только в очереди, объявленной с x-max-priority. В любой другой очереди RabbitMQ его не учитывает. Очередь курьеров pickups объявляет с этим аргументом тот, кто ею владеет. Каждая политика публикует постоянные сообщения, если persistent(false) не задаёт иное, поэтому сообщение в устойчивой очереди переживает перезапуск брокера.

Точка монтирования

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Confirms: a publish resolves once RabbitMQ has taken the message.
        b.include(pack_order)
            .out(DefaultSlot, TransactionalPublish::default())
            .build();
        // Every pickup leaves at priority 3 unless the handler says otherwise.
        b.include(book_courier)
            .out(Pickups, Publish::default().priority(3))
            .build();
    })
}

Вручную

src/main.rs

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Confirms: a publish resolves once RabbitMQ has taken the message.
        b.include(subscriber("orders", PackOrder).build())
            .out(DefaultSlot, TransactionalPublish::default())
            .build();
        // Every pickup leaves at priority 3 unless the handler says otherwise.
        b.include(subscriber(RabbitQueue::new("shipments"), BookCourier).build())
            .out(Pickups, Publish::default().priority(3))
            .build();
    })
}

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

TransactionalPublish::default() - это ConfirmsPublish. Publish::default().priority(3) - обычная политика с приоритетом для каждого сообщения. .confirms() и .server_tx() превращают обычную политику в более строгую и сохраняют её настройки, например обменник. Публикации без подтверждения делят один канал соединения. Издатель с подтверждениями или с транзакциями открывает собственный канал, потому что оба режима - состояние канала.

Сервис публикует и вне обработчика, например из HTTP-обработчика, через издателя, которого приложение выдаёт при запуске (LapinPublish).

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

Тесты в файле примера проверяют, что заказ отправляет все посылки, что заказ с плохой посылкой не отправляет ни одной и какие свойства у каждого вызова курьера. Свойства читаются через tb.out::<Pickups>(). RabbitMQ внутри процесса держит транзакцию в буфере до фиксации и оформляет каждое сообщение так, как это сделал бы сервер. Смысл подтверждения и атомарность транзакции канала определяет сервер, поэтому их проверяет тест на сервере. Обвязку разбирает руководство Тестирование сервиса на RabbitMQ.