发布与应答

本指南用服务发布消息的三种方式向 Pub/Sub 主题发布：应答一条消息、在处理器内部发布，以及在服务就绪后发布一条消息。

每次发布都经过一个发布策略。 Publish 是预导入模块（prelude）中 PubSubPublish 的名字。它向主题发布，并等待服务接受这条消息。主题不存在时发布返回错误：Pub/Sub 不为它没有的主题保留消息。

示例是一个订单服务：它为每个订单应答一张收据，为订单开账单，并通告自己已经启动。

应答

宏
手写

src/main.rs
/// The receipt of an order; its type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

#[subscriber(GooglePubSub::new("orders-receipts").create_with_topic("orders"), reply)]
async fn receipt(order: &Order) -> Receipt {
    tracing::info!(order.id, "receipt sent");
    Receipt { order_id: order.id }
}

src/main.rs
/// The receipt of an order; its type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

struct ReceiptFor;

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

reply 子句让返回值成为一条要发布的消息。它发往哪里，是它的类型的属性：#[outgoing(name = "receipts")] 把每个 Receipt 发往主题 receipts。收据在处理器返回之后、订单确认之前发布。因此收据发布失败时，订单保持未确认，并会回来。

在处理器中发布

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

/// Publishes the invoice of an order to the topic the call names.
#[subscriber(GooglePubSub::new("orders-billing").create_with_topic("orders"))]
async fn bill(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let invoice = Invoice {
        order_id: order.id,
        amount_cents: order.amount_cents,
    };
    match out.message(&invoice).to("invoices").publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, error = %err, "invoice not published, order retried");
            HandlerOutcome::retry()
        }
    }
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

/// Publishes the invoice of an order to the topic the call names.
struct Bill;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Bill
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let invoice = Invoice {
            order_id: order.id,
            amount_cents: order.amount_cents,
        };
        let sent = outs
            .get(DefaultSlot)
            .message(&invoice)
            .to("invoices")
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(order.id, error = %err, "invoice not published, order retried");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Out 参数是处理器自己调用的发布者。 bill 在调用时用 .to("invoices") 指定主题，并决定发布失败对订单意味着什么。这里它用 retry 要求再次投递订单，因此账单不会丢失。处理器在结算之前可以发布任意数量的消息，发往任意主题。

挂载点

宏
手写

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(receipt).out_reply(Publish::default());
            b.include(bill).out(DefaultSlot, Publish::default()).build();
            b.after_startup(Publish::default(), async move |publisher| {
                publisher
                    .message(&Started {
                        service: "orders".to_owned(),
                    })
                    .publish()
                    .await
            });
        },
    )
}

src/main.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-receipts").create_with_topic("orders"),
                    ReceiptFor,
                )
                .reply()
                .build(),
            )
            .out_reply(Publish::default());
            b.include(
                subscriber(
                    GooglePubSub::new("orders-billing").create_with_topic("orders"),
                    Bill,
                )
                .build(),
            )
            .out(DefaultSlot, Publish::default())
            .build();
            b.after_startup(Publish::default(), async move |publisher| {
                publisher
                    .message(&Started {
                        service: "orders".to_owned(),
                    })
                    .publish()
                    .await
            });
        },
    )
}

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

.out_reply(Publish::default()) 为应答指定策略，.out(DefaultSlot, Publish::default()) 为 Out 参数指定它自己的策略。策略保存其位置上每条消息的设置：Publish::default().ordering_key(..) 就是一例。 Broker 为每个主题创建一个客户端发布者，在第一次向该主题发布时创建，此后向该主题的每次发布都复用它。

after_startup 在每个订阅都打开之后运行一次。它收到所指定策略的一个可用发布者，这里用来在 services 上通告服务。钩子返回错误会中止启动，所以无法通告自己的服务不会运行。

运行

服务发布到的主题属于读取它们的一方，所以在服务启动之前就已存在。使用模拟器时，通过它的 REST API 创建这些主题：

for topic in receipts invoices services; do
  curl -X PUT localhost:8085/v1/projects/my-project/topics/$topic
done
cargo run -- run

没有 services 时，通告失败，服务不会启动。

测试

示例文件中的测试发布一个订单，并用 published 断言到达 receipts 和 invoices 的内容。第二个测试启动应用，并在 services 上找到通告。在进程内，每次发布都会被记录，不论是否有订阅读取它的主题。测试套件的用法见测试 Pub/Sub 服务。

应答​

在处理器中发布​

挂载点​

运行​

测试​

应答

在处理器中发布

挂载点

运行

测试