SqsPublish

SqsPublish, Publish in the prelude, sends each message to the queue it names; it is the broker's default, so a reply needs no policy.

A mount site names it wherever a registration publishes: .out(Slot, Publish::default()) for a slot, .out_reply(Publish::default()) for a reply. A lifecycle hook takes it too, b.after_startup(Publish::default(), ..), and is handed the live publisher. The policy holds no connection; the runtime pairs it with the connected broker at startup into an SqsPublisher. A registration that names no policy for its reply publishes through this one, so .out_reply(Publish::default()) only restates the default.

The destination is a queue name or a queue URL. A name resolves to its URL through GetQueueUrl, and the broker caches the answer. A publish never creates a queue: a name the account lacks fails the publish with an error. A publish returns once SQS has accepted the message, or with the error SQS reported.

The payload is the message body, and the headers travel as message attributes. SQS takes a message of up to 1 MiB, body and attributes together. On a .fifo queue every send carries a message group and a deduplication id. A handler sets either for one message through SqsPublishOptions. SnsPublish sends the same messages to an SNS topic instead.

Example

The confirmation names its queue, and the mount site binds no policy for the reply.

Macros
Manual

examples/sqs_publishing_publish.rs
/// The confirmation names its queue, so every reply of this type lands there.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    order_id: u64,
}

#[subscriber(SqsQueue::new("orders"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order_id: order.id }
}

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // No `.out_reply(..)`: the reply rides the broker's default policy, `Publish`.
        b.include(confirm);
    })
}

examples/manual/sqs_publishing_publish.rs
/// The confirmation names its queue, so every reply of this type lands there.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    order_id: u64,
}

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { order_id: order.id }))
    }
}

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // No `.out_reply(..)`: the reply rides the broker's default policy, `Publish`.
        b.include(subscriber("orders", Confirm).reply().build());
    })
}

Example​

Example