跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

用 SNS 扇出

本指南把一条消息向一个 SNS 主题发布一次，让 SNS 把一份副本投递给订阅该主题的每个 SQS 队列。

SNS 到 SQS 是 AWS 服务做扇出的方式：一个主题，每个消费者一个队列。 本 crate 向主题发布；主题背后的每个队列按普通方式读取。 向 SNS 发布需要 sns feature：

Cargo.toml
[dependencies]
ruststream-sqs-sns = { version = "0.7", features = ["sns"] }

示例是一个商店。 它接受订单并在一个主题上发布通知，计费和发货各自从自己的队列读取这条通知。

发往主题的回复​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shop-orders")]
struct PlaceOrder {
    id: u64,
    amount_cents: u64,
}

/// The announcement names the topic, so the handler answers with a bare `reply` clause.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shop-order-events")]
struct OrderPlaced {
    id: u64,
    amount_cents: u64,
}

#[subscriber(SqsQueue::new("shop-orders").create_if_missing(), reply)]
async fn accept(order: &PlaceOrder) -> OrderPlaced {
    tracing::info!(order.id, "order accepted");
    OrderPlaced {
        id: order.id,
        amount_cents: order.amount_cents,
    }
}

accept 用一个 OrderPlaced 回复每个订单，这个类型指定了去向：shop-order-events。 这个名称是队列还是主题，由挂载处决定。

主题背后的队列​

src/main.rs
#[subscriber(SqsQueue::new("shop-billing").create_if_missing())]
async fn bill(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order.id, order.amount_cents, "order billed");
    HandlerOutcome::ack()
}

#[subscriber(SqsQueue::new("shop-shipping").create_if_missing())]
async fn ship(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

bill 和 ship 是普通的队列处理器。 每个队列都得到每条通知的一份副本，所以计费和发货按各自的节奏工作，各自重试。

为队列建立订阅​

src/main.rs
/// Subscribes the billing and shipping queues to the topic, with raw delivery.
///
/// `broker` is a clone of the app's broker, so it uses the connection the app already holds.
async fn subscribe_queues(broker: SqsBroker) -> io::Result<()> {
    let connected = broker.connect().await.map_err(io::Error::other)?;
    for queue in ["shop-billing", "shop-shipping"] {
        connected
            .subscribe_queue_to_topic("shop-order-events", queue)
            .await
            .map_err(io::Error::other)?;
    }
    Ok(())
}

subscribe_queue_to_topic 让一个队列以原始消息传输方式订阅一个主题。 因此副本作为普通的 SQS 消息到达：载荷就是正文，每个消息头就是一个消息属性。 不启用原始消息传输时，SNS 会把消息包装在它自己的 JSON 信封里。 只有队列的访问策略允许主题向它发送时，SNS 才会投递到这个队列。 这个调用在建立订阅之前把这样一条语句加入策略，并保留策略中已有的语句。 主题不存在时，这个调用会创建它；再次为同一个队列建立订阅不会改变任何东西。

拓扑通常属于基础设施，用 Terraform 或控制台创建。 这里由一个启动钩子在应用 Broker 的克隆上完成，克隆使用应用已经持有的连接。

应用​

src/main.rs
fn service(broker: SqsBroker) -> RustStream {
    let topology = broker.clone();
    RustStream::new(AppInfo::new("shop", "0.1.0"))
        .after_startup(async move |_state| subscribe_queues(topology).await)
        .with_broker(broker, |b| {
            // The reply goes to the topic named by `OrderPlaced`, through SNS.
            b.include(accept).out_reply(SnsPublish::default());
            b.include(bill);
            b.include(ship);
        })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

.out_reply(SnsPublish::default()) 把回复绑定到 SnsPublish，回复就变成向主题发出的一次 SNS Publish。 没有这一步时，同样的回复走 Broker 的默认策略 SqsPublish，落到名为 shop-order-events 的队列上。 SnsPublish 用 CreateTopic 按名称打开主题，主题已存在时返回已有的主题；它也接受主题的 ARN 作为名称。 同一个策略可以用在 Out 槽和启动钩子里，ConnectedSqsBroker::sns_publisher 则在应用之外提供一个发布者。

名称以 .fifo 结尾的主题是 FIFO 主题。 它把一次发布的消息组带给订阅它的 FIFO 队列，使用的是 FIFO 队列所用的那些组步骤。 FIFO 队列只能订阅 FIFO 主题。

服务发布需要 sns:CreateTopic 和 sns:Publish 权限。 自己为队列建立订阅时，还需要主题上的 sns:Subscribe，以及每个队列上的 sqs:GetQueueAttributes 和 sqs:SetQueueAttributes。

测试扇出​

示例文件把同一个测试主体运行两次。 在进程内，进程内账户把主题的消息投递给用 subscribe_queue_to_topic 订阅的队列，所以钩子为测试连接拓扑的方式与生产环境相同。 运维人员在服务之外订阅的队列，用 fans_out_to 在策略上指定，这是 testing feature 提供的步骤。 设置了 SQS_TEST_ENDPOINT 时，同一个主体通过 TestApp::start_live 在 LocalStack 上运行：

cargo test --features sns
SQS_TEST_ENDPOINT=http://localhost:4566 cargo test --features sns -- --test-threads=1

测试 SQS 服务介绍测试工具和线上模式。