SqsPublishOptions

A handler sets one message's group or deduplication id through the publish steps, bounding its slot on SqsPublishOptions.

SqsPublishOptions holds the two settings one publish may differ from the next in: a message group and a deduplication id. A call site writes them through the steps of SqsPublishSteps, .group_id(..) and .deduplication_id(..) on the publish builder. A step wins over the group the policy fixed, for that one message. Both are FIFO settings: a .fifo destination takes them, and a standard one fails the publish.

A handler bounds its slot as Out<impl Publisher<Options = SqsPublishOptions>, Marker>, which puts the steps in reach. The body imports ruststream_sqs_sns::prelude::*, the one handler file that globs this crate's prelude instead of the framework's. The same steps are on every publish surface of this crate: an injected slot, the publisher a lifecycle hook is handed, a fan-out publisher. A slot moved from SqsPublish to SnsPublish keeps the body that names them. A reply has no call site, so its policy is its whole answer. A test reads back what a publish asked for: tb.out::<Marker>().with_options(&SqsPublishOptions::default().group_id(..)).

Example

Each customer's orders reach fulfillment in the order they were placed: the group of each message is the customer.

Macros
Manual

examples/sqs_fifo_group.rs
/// One customer's orders reach fulfillment in the order they were placed; different customers'
/// orders do not wait for each other. The group is the customer.
#[subscriber(SqsQueue::new("orders"))]
async fn place(
    order: &Order,
    Out(placements): Out<impl Publisher<Options = SqsPublishOptions>, Placements>,
) -> HandlerOutcome {
    let placed = OrderPlaced {
        id: order.id,
        customer: order.customer.clone(),
    };
    let sent = placements
        .message(&placed)
        .group_id(&order.customer)
        .publish()
        .await;
    if let Err(err) = sent {
        tracing::warn!(order.id, %err, "order not placed");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/sqs_fifo_group.rs
/// One customer's orders reach fulfillment in the order they were placed; different customers'
/// orders do not wait for each other. The group is the customer.
struct Place;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Place
where
    Entry: OutEntry<Placements, Wire: Publisher<Options = SqsPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let placed = OrderPlaced {
            id: order.id,
            customer: order.customer.clone(),
        };
        let sent = outs
            .get(Placements)
            .message(&placed)
            .group_id(&order.customer)
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(order.id, %err, "order not placed");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Example​

Example