共享状态与上下文

支付客户端和邮件发送器在启动时打开一次，以 State 的形式到达每个订阅者；上下文告诉订阅者它拿到的是哪次投递。

服务在订阅者之间共享客户端、连接池和缓存。应用在 on_startup 中打开它们，时机在第一个订阅打开之前。这里出现错误会中止启动。钩子的返回值是应用的状态，整个服务只有这一个值。订阅者以 State<T> 参数的形式取用其中的一个字段。上下文是按投递划分的那一半：订阅的名字、消息头、Broker 为这次投递提供的字段。订阅者以 &mut Context 参数的形式取用它，紧跟在消息之后。

状态

Shop 持有支付客户端和邮件发送器，两者各自放在 Arc 后面，所以克隆的开销很小。 FromRef 从状态中投影出每个字段：State<Arc<Gateway>> 就是通过它解析的。 open 是第 5 章中应用传给 on_startup 的钩子。

宏
手写

state.rs
/// The app's state: each field reaches a subscriber as `State<FieldType>`.
#[derive(Clone, FromRef)]
pub(crate) struct Shop {
    pub(crate) gateway: Arc<Gateway>,
    pub(crate) mailer: Arc<Mailer>,
}

impl Shop {
    /// Opens the clients. The app runs it once, before the first subscription opens; an error
    /// stops the startup.
    pub(crate) async fn open() -> Result<Self, MailError> {
        Ok(Self {
            gateway: Arc::new(Gateway::new(CARD_LIMIT_CENTS, CHARGES_PER_SECOND)),
            mailer: Arc::new(Mailer::connect("smtp://mail.internal:25").await?),
        })
    }
}

state.rs
/// The app's state: each field reaches a subscriber as `State<FieldType>`.
#[derive(Clone)]
pub(crate) struct Shop {
    pub(crate) gateway: Arc<Gateway>,
    pub(crate) mailer: Arc<Mailer>,
}

// What `#[derive(FromRef)]` writes for each field: `State<Arc<Gateway>>` resolves through it.
impl FromRef<Shop> for Arc<Gateway> {
    fn from_ref(shop: &Shop) -> Self {
        shop.gateway.clone()
    }
}

impl FromRef<Shop> for Arc<Mailer> {
    fn from_ref(shop: &Shop) -> Self {
        shop.mailer.clone()
    }
}

impl Shop {
    /// Opens the clients. The app runs it once, before the first subscription opens; an error
    /// stops the startup.
    pub(crate) async fn open() -> Result<Self, MailError> {
        Ok(Self {
            gateway: Arc::new(Gateway::new(CARD_LIMIT_CENTS, CHARGES_PER_SECOND)),
            mailer: Arc::new(Mailer::connect("smtp://mail.internal:25").await?),
        })
    }
}

手写形式写出 derive 生成的两个 impl，每个字段一个。

同一个订阅者中的状态与上下文

send 通过状态中的邮件发送器用电子邮件发出通知。不存在的邮箱再试也还是不存在，所以被拒绝的通知直接丢弃，警告中指明取自上下文的订阅。

宏
手写

notifications.rs
/// E-mails a notice to the customer.
// The mailer comes from the app's state; the context names the subscription in the warning.
#[subscriber("notices")]
async fn send(
    notice: &Notice,
    ctx: &mut Context<'_, (), Shop>,
    State(mailer): State<Arc<Mailer>>,
) -> HandlerOutcome {
    match mailer.send(&notice.customer, &notice.text).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            // A refused mailbox stays refused; another try would only repeat the warning.
            tracing::warn!(subscription = ctx.name(), %error, "notice not sent");
            HandlerOutcome::drop()
        }
    }
}

notifications.rs
/// E-mails a notice to the customer.
struct Mail;

// The mailer comes from the app's state; the context names the subscription in the warning.
impl Handle<Notice, (), (), (), Shop> for Mail {
    async fn handle(
        &self,
        notice: &Notice,
        _outs: &(),
        ctx: &mut Context<'_, (), Shop>,
    ) -> Result<(), HandlerOutcome> {
        let State(mailer) = State::<Arc<Mailer>>::from_context(ctx).await?;
        if let Err(error) = mailer.send(&notice.customer, &notice.text).await {
            // A refused mailbox stays refused; another try would only repeat the warning.
            tracing::warn!(subscription = ctx.name(), %error, "notice not sent");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

在手写形式中，状态类型是 Handle 的最后一个参数，挂载时会把它与应用的状态类型核对。主体用 from_context 读取字段，这正是宏对 State 参数所做的事。自定义提取器是你自己的类型，以同样的方式读取。

状态​

同一个订阅者中的状态与上下文​

状态

同一个订阅者中的状态与上下文