服务和它的第一个订阅者

商店从一个订阅者起步。它接收每一个已下的订单，由一个小应用挂载在内存 Broker 上。

RustStream 服务是一个应用：一个构建器，它列出自己的 Broker，并在上面挂载订阅者。 订阅者是一个 async fn，第一个参数是消息，#[subscriber] 指明它读取的主题。 运行时在主体运行之前解码载荷，默认使用 JSON。 主体的返回值决定如何结算这次投递：HandlerOutcome::ack 告诉 Broker 这条消息已处理完毕。 内存 Broker 在进程内部传递消息，所以商店不需要安装任何东西。 第 11 章把它换成联网的 Broker，订阅者保持不变。

订单

订单是一个 serde 类型。 Outgoing derive 声明订单发布到哪里：orders.placed。 商店自己还不发布订单。现在由测试发布，第 10 章起由 HTTP 店面发布。

宏

examples/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders.placed")]
struct Order {
    id: u64,
    customer: String,
}

手写

examples/manual/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Order {
    id: u64,
    customer: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "orders.placed")]` writes.
impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders.placed";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

手写形式写出 derive 生成的两个 impl：目的地，以及订单携带的消息头（目前没有）。

第一个订阅者

accept 读取 orders.placed，并通过 tracing 记录每个订单。 手写形式是一个具名类型，带有 impl Handle<Order>：Ok(()) 表示 ack，Err(outcome) 按该结果结算。

宏

examples/guide/first.rs

/// Takes every placed order; for now it only logs it.
#[subscriber("orders.placed")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order = order.id, customer = %order.customer, "order accepted");
    HandlerOutcome::ack()
}

手写

examples/manual/guide/first.rs

/// Takes every placed order; for now it only logs it.
struct Accept;

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

应用

RustStream 就是应用。with_broker 注册一个 Broker，并用 include 在上面挂载订阅者。 构建器只记录交给它的内容：Broker 在服务启动时才连接。 #[ruststream::app] 属性把构建器变成 main，并带有 run 和 asyncapi gen 命令。 手写形式亲手写出这个 main：在构建器之上调用 run_main。

宏

examples/guide/first.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
    })
}

手写

examples/manual/guide/first.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders.placed", Accept).build());
    })
}

/// What `#[ruststream::app]` writes: the `run` and `asyncapi gen` commands over `app`.
fn main() -> ExitCode {
    run_main(app)
}

运行

cargo run --example guide_first --features testing,macros,memory,json,logging -- run

logging feature 把服务的日志打印到终端。 服务连接 Broker，打开订阅，然后一直运行到 Ctrl+C：

INFO ruststream::lifecycle starting service service=shop version=0.1.0 brokers=1 subscribers=1
INFO ruststream::lifecycle broker connected broker="ruststream::memory::MemoryBroker"
INFO ruststream::dispatch subscriber started subscriber=orders.placed input="guide_first::Order"
INFO ruststream::lifecycle service running subscribers=1

本指南的其余部分把这个文件逐步扩展成 examples/guide 中的商店。