编写一个 Broker¶

英文版本为准

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

Broker 是一个实现了核心 trait 的独立 crate。它依赖 ruststream 并关闭默认 feature，因此只拿到 trait 接口和运行时，不带内置的 JSON 编解码器，也不带别的 Broker：

[dependencies]
ruststream = { version = "0.7", default-features = false }

本页就是这份契约。实现必需的 trait，定义你自己的 Config，为你的 Broker 支持的功能实现能力 trait，再用 conformance 校验套件验证结果。基于真实客户端的完整实现见 NATS 完整示例。

必需的 trait¶

`Broker` 与 `ConnectedBroker`¶

Broker 只负责生命周期。生命周期是一串状态转移：每个状态都是不同的类型，转移消费掉当前状态并交出下一个状态，因此顺序错乱的调用无法通过编译。Broker 不指定订阅者类型，也不指定发布者类型，所以一个应用可以混用不同种类的 Broker。

pub trait Broker: Send + Sync + Sized {
    type Error: std::error::Error + Send + Sync + 'static;
    type Connected: ConnectedBroker;
    async fn connect(self) -> Result<Self::Connected, Self::Error>;
}

pub trait ConnectedBroker: Send + Sync + Sized + 'static {
    type Error: std::error::Error + Send + Sync + 'static;
    type Closed: Send;
    async fn shutdown(self) -> Result<Self::Closed, Self::Error>;
}

shutdown 不得阻塞，也不得 panic。所有可能返回错误的资源释放都在这里做完，并返回 Result。Closed 是关闭的见证：把清理时的诊断信息（缓冲区刷新结果、丢弃计数）当作普通数据放进去，或者直接用 ()。

构造是同步且不做 I/O 的：new(addrs) 只记录配置。所有网络工作都发生在 connect 里，运行时在启动时调用它一次。已连接形态直接持有活的客户端，因此它自身的操作不必检查“是否已经连接”。

Broker 还可以额外持有一个由 connect 填充的共享单元，或者像内存 Broker 那样持有可共享的进程内状态。这样在应用还在组装、connect 还没运行的时候，就可以先把发布者交出去：该单元服务的是这些提前拿到的句柄，而不是已连接形态。

conformance 校验套件会验证整条转移链，NATS 示例则在真实客户端上走完这条链。

在一个已经关闭的 Broker 上，没有发布或订阅方法可以调用，所以持有者一侧的误用通不过编译。共用连接只能在运行时检查：共用它的句柄（已连接形态交出去的发布者、可共享 Broker 的克隆）在关闭之后使用时必须返回错误，绝不能悄悄返回成功。lifecycle 检查也会走到这条路径。

内存 Broker 用几行就走完整条转移链。本页下面的每一段示例也都出自同一个文件，契约一改，页面上的代码就跟着改：

src/memory/mod.rs

impl<Log: LogMode> Broker for MemoryBroker<Log> {
    type Error = MemoryError;
    type Connected = ConnectedMemoryBroker<Log>;

    /// Connecting is free for an in-process bus. A shut-down bus (a clone lineage may have shut
    /// the shared state down) is revived with a fresh, empty registration map, so the connected
    /// form always starts live; a live bus keeps its registrations.
    fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> {
        {
            let mut bus = self
                .state
                .subscribers
                .lock()
                .expect("memory broker mutex poisoned");
            if matches!(*bus, Bus::ShutDown) {
                *bus = Bus::Live(HashMap::new());
            }
        }
        ready(Ok(ConnectedMemoryBroker {
            state: self.state,
            mode: PhantomData,
        }))
    }
}

/// The connected form of [`MemoryBroker`]: the typed witness that [`Broker::connect`] ran.
///
/// Cheap to clone: the in-memory bus is shared state by nature, so the connected form is a
/// shareable handle on it, exactly like the unconnected broker. Subscriptions (the
/// [`Subscribe`] capability, [`MemorySource`]) resolve against this form, and carry over the
/// broker's log mode: only a [`Retaining`] one opens repositionable subscriptions.
pub struct ConnectedMemoryBroker<Log = Discarding> {
    state: Arc<MemoryState>,
    mode: PhantomData<Log>,
}

impl<Log> Clone for ConnectedMemoryBroker<Log> {
    fn clone(&self) -> Self {
        Self {
            state: Arc::clone(&self.state),
            mode: PhantomData,
        }
    }
}

impl<Log> fmt::Debug for ConnectedMemoryBroker<Log> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ConnectedMemoryBroker")
            .finish_non_exhaustive()
    }
}

impl<Log: LogMode> ConnectedMemoryBroker<Log> {
    /// Returns a publisher bound to this broker.
    #[must_use]
    pub fn publisher(&self) -> MemoryPublisher {
        MemoryPublisher {
            state: Arc::clone(&self.state),
            txn: Mutex::new(None),
        }
    }

    /// Returns a request / reply-capable publisher bound to this broker.
    ///
    /// See [`MemoryBroker::requester`] for why its operations report [`RequestError`] rather
    /// than [`MemoryError`].
    #[must_use]
    pub fn requester(&self) -> MemoryRequester {
        MemoryRequester::new(Arc::clone(&self.state))
    }
}

impl<Log: LogMode> ConnectedBroker for ConnectedMemoryBroker<Log> {
    type Error = MemoryError;
    type Closed = ClosedMemoryBroker;

    /// Enters the terminal shut-down state: the bus itself flips to its `ShutDown` variant, so
    /// every aliased handle that would touch it (a publisher's publish or transaction commit, a
    /// request) errors with [`MemoryError::ShutDown`]. Consuming `self` makes any further use
    /// of this handle a compile error; the returned witness reports how many subscriber
    /// registrations the teardown dropped.
    fn shutdown(self) -> impl Future<Output = Result<Self::Closed, Self::Error>> {
        let dropped = {
            let mut bus = self
                .state
                .subscribers
                .lock()
                .expect("memory broker mutex poisoned");
            match std::mem::replace(&mut *bus, Bus::ShutDown) {
                Bus::Live(subscribers) => subscribers.values().map(Vec::len).sum(),
                Bus::ShutDown => 0,
            }
        };
        ready(Ok(ClosedMemoryBroker {
            subscribers_dropped: dropped,
        }))
    }
}

ClosedMemoryBroker 就是上面说的那种带清理诊断的见证：它报告这次关闭移除了多少个订阅者注册。

`Subscribe`¶

在已连接形态上实现 Subscribe，服务就能按主题、subject 或队列的名字订阅。#[subscriber("name")] 用的就是它。

pub trait Subscribe: ConnectedBroker {
    type Subscriber: Subscriber;

    // 按名字订阅时，重试副本由谁发布、由谁说出它的地址。按订阅名发布就能到达用这个
    // 名字打开的订阅时（subject、topic、流、队列名通常如此）答 AddressedCopies，
    // 这时名字本身就是地址。不是这样就答 NamedCopies：MQTT 的过滤器读很多 topic，
    // 一个也说不出来。
    type Copies: CopyPath;

    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>;

    // 默认实现：用这个名字挂载的注册对自己的重试声明了什么。如果 Broker 有对应的机
    // 制，就把上限和地址映射到这个名字打开的那条订阅上。
    fn declare_retry(&self, name: &str, declaration: &RetryDeclaration)
        -> Result<(), DeclareRetryError>;
}

要做的只有建立一条订阅，再说出发布按哪个地址能重新到达它：

src/memory/mod.rs

impl<Log: LogMode> Subscribe for ConnectedMemoryBroker<Log> {
    type Subscriber = MemorySubscriber<Log>;
    // One subject is both ends of the bus here, so a publish under the name a subscription reads
    // reaches that subscription, and the name is the address.
    type Copies = AddressedCopies;

    fn subscribe(&self, name: &str) -> impl Future<Output = Result<Self::Subscriber, Self::Error>> {
        let (tx, rx) = mpsc::unbounded_channel();
        let name = name.to_owned();
        if let Err(err) = self.state.register(&name, tx.clone()) {
            return ready(Err(err));
        }
        ready(Ok(MemorySubscriber {
            name,
            rx,
            requeue: tx,
            state: Arc::clone(&self.state),
            seek: Arc::new(SeekControl::default()),
            mode: PhantomData,
        }))
    }
}

type Copies 是按名字订阅的源在你的 Broker 上报出的答案，#[subscriber("orders")] 怎么重试由它决定：答 AddressedCopies 时名字本身就是地址，别的什么也不用写；答 NamedCopies 时副本发往哪里由挂载处说出来。

订阅名不是发布地址的地方，答 NamedCopies。MQTT 的主题过滤器就是这样：devices/+/telemetry 读每台设备的 topic，却一个也没有说出来，于是副本发往哪里由挂载处说出 - 用 .out_retry(policy).to(name) 固定下来，或者用发布变换为每次投递各自命名。只有一个名字还不够的订阅，要用你自己的描述符。

`Subscriber`¶

订阅者是入站消息的 Stream。背压由 Stream 本身给出。

pub trait Subscriber: Send {
    type Message: IncomingMessage;
    type Error: std::error::Error + Send + Sync + 'static;
    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_;
}

stream 取 &mut self，因此两次 poll 之间缓冲的状态都存放在这个可变借用后面，取消安全正来自这一点。

`IncomingMessage`¶

一条投递过来的消息交出自己的载荷和消息头，并由 ack 确认或由 nack 退回。ack 消费 self，因此两次 ack 是编译错误。

这里的每个访问方法交出的都是借用，因此一条投递只需要一个引用计数，而不是每个字段各有一个。Broker 的投递包住客户端自己的消息，而它的名字、载荷和消息头通常是三个各自计数的值；把它们放进一个 Arc 后面的块里，把每份副本各不相同的东西留在块外，例如日志位置和投递次数。这样一次交接（发给每个订阅者的副本、重新入队、重放）就是一次原子自增而不是三次。客户端线程和分发任务位于不同核心的服务，每条消息都在付这笔钱。MemoryBroker 就是这样写的。

pub trait IncomingMessage: Send + Sync {
    fn payload(&self) -> &[u8];
    fn headers(&self) -> &HeaderMap;
    async fn ack(self) -> Result<(), AckError>;
    async fn nack(self, requeue: bool) -> Result<(), AckError>;

    // Defaulted: false. The runtime reads this first and never calls
    // nack_after without it, so override the pair together.
    fn supports_nack_after(&self) -> bool;

    // Defaulted: AckError::Unsupported. Override when the transport has native
    // delayed redelivery (JetStream NAK with delay); handlers reach it through
    // HandlerOutcome::retry_after.
    async fn nack_after(self, delay: Duration) -> Result<(), AckError>;

    // Defaulted: None. Override (with the Partitioned capability) to feed the
    // runtime's keyed worker lanes, workers(n, by_key).
    fn partition_key(&self) -> Option<&[u8]>;

    // Defaulted: None. Override where the transport counts its own deliveries
    // (JetStream num_delivered, SQS ApproximateReceiveCount, Pub/Sub
    // delivery_attempt): a registration's max_attempts(..) cap then counts the
    // broker's redeliveries and not only the copies the runtime published.
    // The first delivery of a message answers 1.
    fn redelivery_count(&self) -> Option<u64>;
}

延迟重新投递由两个方法组成，运行时问的是 supports_nack_after。只覆盖 nack_after，这个标志仍然是 false，运行时一次也不会调用这个覆盖。nack_after 的默认实现返回 AckError::Unsupported，而不是按一次普通的 nack(true) 结算：留不住消息的传输必须说出这一点，否则一次退避就变成一场重新投递的风暴。

这四个带默认实现的方法一个都不覆盖的 Broker，仍然能配合运行时的每一项功能。没有原生延迟重新投递的地方，retry_after 由运行时自己完成：它丢弃这次投递，并在延迟之后经由这条注册的重试发布者发布一份副本，同时把重试计数消息头加一。这份副本发往你的订阅给出的地址。按键分道的工作者池轮流分发没有键的消息。

只要 redelivery_count 还返回 None，这个消息头就是唯一的计数，max_attempts(..) 的上限也按它来读。覆盖了这个方法，Broker 自己的计数就成了上限唯一读的那个：框架的消息头不会再加进来。

延迟由你自己发布一份副本来兑现时 - 死信交换机后面的一条等待队列，或者一个专门的重试主题 - 请在那份副本上把 RETRY_COUNT_HEADER（从 ruststream::runtime 导出）加一，但只在传输什么都不记的时候这样做。传输自己记次数时，你发布的副本是一条新消息，Broker 从一开始重新记，于是绕完等待队列回来的那次投递，在上限看来仍是第一次尝试。这是你这套延迟方案的行为：把它写进 crate 的文档，别去动那个消息头。

“什么都不覆盖”会得到什么，没有哪个 Broker 可以拿来演示：这个工作区里的 Broker 个个都覆盖了这三个方法。所以这份行为由核心的一个测试固定下来：

src/message.rs

struct Stub {
    payload: Vec<u8>,
    headers: HeaderMap,
}

impl IncomingMessage for Stub {
    fn payload(&self) -> &[u8] {
        &self.payload
    }

    fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    fn ack(self) -> impl Future<Output = Result<(), AckError>> {
        ready(Ok(()))
    }

    fn nack(self, _requeue: bool) -> impl Future<Output = Result<(), AckError>> {
        ready(Ok(()))
    }
}

let stub = Stub {
    payload: b"body".to_vec(),
    headers: HeaderMap::new(),
};
assert_eq!(stub.payload(), b"body");
// The default partition_key is None (no key).
assert!(stub.partition_key().is_none());
// The default reports no native delayed redelivery, so the runtime uses its fallback.
assert!(!stub.supports_nack_after());
// The default nack_after signals "not honored" rather than silently degrading.
assert!(matches!(
    stub.nack_after(Duration::from_secs(1)).await,
    Err(AckError::Unsupported)
));

正是 Unsupported 这个答复让运行时分得清两种情况：传输没有延迟投递，还是延迟已经生效。分清之后，它才走自己的备用路径。

`Publisher`¶

pub trait Publisher: Send + Sync {
    /// 你的传输层怎样消费载荷：只读它就是 `Lend`，你的客户端要留着它就是 `Take`。
    type Payload: PayloadForm;

    type Error: std::error::Error + Send + Sync + 'static;

    /// 你的 Broker 的逐条消息设置。每个字段都是可选的；没有这类设置就写 `()`。
    type Options: Clone + Send + Sync + 'static;

    async fn publish(
        &self,
        msg: OutgoingFor<'_, Self::Payload>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error>;

    /// 带默认实现：这个发布者垫在每次发布下面的消息头。
    fn base_headers(&self) -> Option<&HeaderMap> { None }
}

客户端在调用结束之后还留着载荷 - 它收下的是 Vec<u8>、Bytes 或别的自有值 - 就声明 Take；传输层只是把字节写进自己的帧、批或者套接字缓冲区，就声明 Lend。几乎每一种线上协议都属于后者。

你收到的消息跟着这个声明走，框架在你上面做的事情也一样：

// Take：框架写好的那个缓冲区，归你。
async fn publish(&self, msg: OutgoingMessage<'_, BytesMut>, options: Option<&()>) -> Result<(), Error>

// Lend：字节就在它原来的地方，在这次调用期间有效。
async fn publish(&self, msg: OutgoingMessage<'_, &[u8]>, options: Option<&()>) -> Result<(), Error>

声明 Take 的发布者拿到的是编解码器自己的缓冲区，写成什么样就是什么样。Vec::from(payload) 不花代价，因为那个缓冲区本身就是向量；payload.freeze() 花的是让所有权可共享的那一个内存块；只有借来别人字节的那种发布才会在进来的路上复制一次。

声明 Lend 的发布者拿到的是 &[u8]，没有什么需要释放。在派发循环里，框架把一个循环一份的缓冲区借给它，每条消息都清空重写，所以经由你这个 Broker 的回复在每次投递上一次内存分配都没有。这正是这个声明存在的理由：运行时不能把你可能留下的东西借出去。

OutgoingMessage 借用自己的名字，载荷和消息头映射则是你的传输层可以接管而不必复制的东西。读用 msg.payload() 和 msg.headers()，无论你声明的是哪种形态，它们都给出 &[u8] 和 &HeaderMap。接管整条消息的传输层用 msg.into_parts() 取走它的各个部分 - 目的地、载荷和映射一次移动交出，什么都不复制。

服务写的不是这个方法，而是构建器：publisher.message(&value).publish() 选定目的地、编解码器和消息头，然后恰好调用一次 publish。实现 publish，整个构建器就在它之上工作起来。

Options 装的是属于消息而不属于句柄的东西：QoS、优先级、排序键、过期时间。一次调用只带上它改动过的字段，其余的就是策略在构造这个发布者时定下的值。把两者合起来是你的 publish 要做的第一件事。

在没有调用点可以改动设置的路径上，options 是 None：处理器的回复、延后重试。那里生效的是策略的设置。

Clone 和 'static 是测试套件对这个类型的要求：它把经由 Out 槽位的一次发布所带的设置复制一份，再以这个类型交回给测试。于是，测试你这个 Broker 的服务断言的是你的 publish 收到的值，而不是它变成的那个协议字段。再派生 Debug 和 PartialEq，断言就写成 with_options(&YourOptions { .. }) （见对 Out 槽位做断言）。

base_headers 留给发布者自身的常量：租户、producer 名字、这个句柄每条消息都带的 schema id。构建器以这份基础消息头为起点，再把调用点的消息头逐个键写在上面，所以同一个键上留下的是调用点的值（参见消息头从哪里来）。

Transaction 指定自己的 Options 和 Payload，也带同样的默认 base_headers。事务是一个独立的发布面，所以它认可的设置可以和开启它的发布者不同，而客户端缓冲区会留着载荷，直接发布却只是读它。多数 Broker 在这两处写的都是发布者自己的类型。自身没有常量的句柄，两处的 base_headers 都停在默认实现上。

`PublishPolicy`¶

Broker 的发布者由两部分组成：一份策略（一个 exchange、一个队列超时、一个事务 id）和一条活连接。提供一个单独的策略类型：它在任何地方都能构造，持有构建器选项。

在它上面实现 PublishPolicy：策略在已连接形态上构造出活的发布者，而 pair 就是这个构造函数。它是异步的，也可以返回错误，需要初始化事务性 producer 的 Broker 就在这里做这件事。

pub trait PublishPolicy<C: ConnectedBroker> {
    type Live; // the live publisher (or live wiring form, for combinator stacks)
    async fn pair(self, connected: &C) -> Result<Self::Live, PairError>;
}

错误类型是做了类型擦除的 PairError：用 PairError::new 包住你的 Broker 自己的错误。策略在启动时把发布者实例化一次，因此 pair 不会落到热路径上。

为每一种真正意义上的发布模式提供一对策略和活形态，并且让模式的选择成为策略类型之间的转移，而不是一个运行时标志。普通策略构造出普通的发布者，而 transactional_id(..) 这一步构建器调用把它转成另一个独立的事务性策略类型，它的活形态实现 TransactionalPublisher。于是普通发布者上根本没有事务接口。

最小的参考实现是内存 Broker 的 MemoryPublish 和 MemoryRequest：它们没有选项，所以是空结构体。

核心的类型化组合子以函子的方式实现 PublishPolicy，所以用户可以在策略构造出发布者之前，先在它之上组合编解码器和变换。

如果普通策略用自己的默认值就够用（几乎总是如此），就在已连接形态上再实现 DefaultPublish，在那里指名这个策略。这样，带 publish("dest") 的处理器在没有显式 .out_reply(..) 的情况下挂载，运行时自己就把回复用的发布者实例化出来，只写 b.include(def) 也能编译。发布者总是需要显式选项的 Broker 不实现 DefaultPublish，它们的用户在每次注册处理器时指定策略。

pub trait DefaultPublish: ConnectedBroker {
    type Policy: PublishPolicy<Self> + Default + Send + 'static;
}

下面是这两半，取自一个策略完全没有选项的 Broker：

src/memory/mod.rs

impl<Log: LogMode> PublishPolicy<ConnectedMemoryBroker<Log>> for MemoryPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker<Log>,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(connected.publisher()))
    }
}

impl<Log: LogMode> DefaultPublish for ConnectedMemoryBroker<Log> {
    type Policy = MemoryPublish;
}

订阅来源¶

Subscribe 覆盖的是一个名字就够用的情形。订阅需要你的 Broker 专有的选项（消费者组、持久化名称、投递策略）时，定义一个实现 SubscriptionSource 的描述符类型：

pub trait SubscriptionSource<C: ConnectedBroker> {
    type Subscriber: Subscriber;

    // 这条订阅的重试所用的副本由谁发布、由谁说出它的地址：AddressedCopies、
    // NamedCopies 或 BrokerMoves。
    type Copies: CopyPath;

    fn name(&self) -> &str;
    fn subscribe(self, connected: &C) -> impl Future<Output = Result<Self::Subscriber, C::Error>> + Send;

    // 默认返回原样的描述符。在这里读出这条注册的 max_attempts(..) 和
    // dead_letter(..)，如果 Broker 有对应的机制，就把它们用到即将打开的订阅上。
    fn declare_retry(self, declaration: &RetryDeclaration) -> Self;

}

// AddressedCopies 的另一半：地址是类型的属性，而不是启动时才检查的答案。只有活连接
// 知道时，就去问 Broker。
pub trait RedeliveryAddressed<C>: SubscriptionSource<C, Copies = AddressedCopies> {
    async fn redelivery_address(&self, connected: &C) -> Result<RedeliveryAddress, C::Error>;
}

给描述符一个关联构造函数（OrdersStream::new(..)），而不是自由函数：这样用户就能在属性里直接写出它，#[subscriber(OrdersStream::new("orders", "workers"))]。

宏从这次构造调用里读出类型，只要每个方法都返回 Self，也接受在它之上的构建器链（#[subscriber(OrdersStream::new("orders").durable("workers"))]）。

type Subscriber 声明在来源上，所以一个 Broker 可以提供多种订阅方式（pub/sub 和流），各带不同的订阅者类型；也可以像 NATS 示例那样，用一个在内部分支的描述符服务全部方式。

给描述符派生 Clone：它是配置，挂载点为每次注册重新构造它，所以同一个定义可以同时挂到两个 Broker 上。

谁来发布重试副本¶

每个描述符用 type Copies 在三者中声明一个。

AddressedCopies 表示重试需要的副本由本进程发布，而且描述符知道它们发往哪里。它在 SubscriptionSource 旁边再实现 RedeliveryAddressed，于是地址是类型的属性，而不是启动时才检查的答案。subject、topic、流和队列的答案都是它：一条订阅，一个服务自己能发布回去的地址。

NamedCopies 表示副本由本进程发布，但描述符给不出地址：带通配符的 subject、MQTT 的过滤器、 Pulsar 的 pattern、一串 topic。这样的订阅读很多地址，于是由挂载处说出一个 - 用 .out_retry(policy).to(name) 固定下来，或者用发布变换为每次投递各自命名。

BrokerMoves 表示投递由服务端或客户端库自己搬走：带 x-delivery-limit 和 x-dead-letter-exchange 的 quorum 队列、带死信策略的 Pub/Sub 订阅、SQS 的 redrive 策略、带 DeadLetterPolicy 的 Pulsar 消费者。这时服务本身什么也不发布，因此在这个描述符的任何挂载处写 .out_retry(..) 都是编译错误，错误会报出这个描述符。

前两者都会为每条注册从 Broker 的 DefaultPublish 策略绑定一个重试发布者，所以在没有 DefaultPublish 的 Broker 上声明它们中的任何一个，代码都编译不过。

Subscribe 为按名字订阅的形式声明同一件事：那里的 type Copies 就是 #[subscriber("orders")] 在你的 Broker 上报出的答案。按订阅名发布就能到达用这个名字打开的订阅时答 AddressedCopies - subject、topic、流和队列名通常如此 - 这时名字本身就是地址，别的什么也不用写。

原生与否取决于字段的取值而不是类型时 - 没有 .delay(..) 的 RabbitMQ 队列并没有自己的延迟投递 - 把这条路径保持开放。

注册声明了什么¶

declare_retry 把挂载处声明的上限和地址交给你，每条注册一次，并且在 subscribe 之前。自带机制的描述符在这里把它们变成拓扑，而且只在两者都声明时才这么做，因为原生的死信策略同时需要上限和地址。没有这种机制的描述符保留默认实现，声明改由运行时在重试路径上落实。

用一个光名字挂载的注册声明的是同一件事，而它拿到的描述符是核心的 Name，里面没有可以安放声明的拓扑。这时改由你的 Broker 在 Subscribe::declare_retry 里接住它：时机相同，对象是它即将打开的那个名字。在那里按描述符的做法映射它，并且只在两半都声明时才映射。

默认实现接受什么也没声明的注册。只要你的 type Copies 说副本由本进程发布，它也接受任何声明：上限和地址由运行时落实。在 BrokerMoves 的 Broker 上，它在启动时拒绝非空的声明，并报出这条订阅和声明该写的地方：再没有别人会落实它，而上限悄悄消失的消息会活过自己的死信策略。光名字能够到的机制，就把这个方法实现出来 - Pub/Sub 的死信策略、SQS 的 redrive 策略、Pulsar 消费者的 DeadLetterPolicy - 其余情况不要碰它。

重试副本发往哪里¶

没有原生延迟重新投递时，运行时自己兑现 retry_after：等延迟过去，它发布一份消息的副本。副本发往哪里，由声明了 AddressedCopies 的描述符说出来。

src/memory/mod.rs

impl<Log: LogMode> SubscriptionSource<ConnectedMemoryBroker<Log>> for MemorySource {
    type Subscriber = MemorySubscriber<Log>;
    // The bus moves nothing on its own, and one subject is both ends of it, so a copy goes back
    // to the subject the subscription reads.
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        &self.name
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker<Log>,
    ) -> Result<Self::Subscriber, MemoryError> {
        Subscribe::subscribe(connected, &self.name).await
    }
}

impl<Log: LogMode> RedeliveryAddressed<ConnectedMemoryBroker<Log>> for MemorySource {
    fn redelivery_address(
        &self,
        _connected: &ConnectedMemoryBroker<Log>,
    ) -> impl Future<Output = Result<RedeliveryAddress, MemoryError>> + Send {
        // One subject is both ends of the bus, and no lookup is needed to say so.
        ready(Ok(RedeliveryAddress::new(self.name.clone())))
    }
}

返回的名字，要让指向你的 Broker 的发布者用它就能重新到达这条订阅：NATS 上是 subject，Kafka 上是 topic，Redis 上是流的键。

在 NATS JetStream 上，消费者绑定的是流而不是 subject，所以答案是这个流所发布的某个 subject，由流名构造出来的描述符要向服务端问出来。运行时只在启动时问一次，挂载处的 .to(name) 会覆盖这个答案。

.to(name) 命名的是这条注册要发往的通道，所以生成的文档会把它连同一个 send 操作一起报出来。你回答的那个地址不会被报出来：它是这条订阅自己的通道，文档里已经有了。

harness::redelivery_address 会按你给出的答案检查：发往所报地址的一次发布，必须到达报出它的那条订阅。声明了 NamedCopies 的描述符没有可检查的答案，两者其余的转移链都由 harness::lifecycle 覆盖。

用一个字符串命名一种订阅方式¶

只由一个名字确定、再没有别的标识的订阅方式，还会实现 FromName：它唯一的构造函数用这个名字构造出值。

src/memory/mod.rs

impl FromName for MemorySource {
    fn from_name(name: impl Into<Cow<'static, str>>) -> Self {
        Self::new(name.into().into_owned())
    }
}

于是 #[subscriber(OrdersStream)] 就合法了：属性指定订阅方式，值由挂载点补上。确实需要不止一个名字才能成立的方式（既要一个主题，又要一个订阅名）不实现 FromName，这种写法对它就通不过编译。

用你自己的词汇表达配置¶

核心不知道订阅还有流、持久化名称或消费者组，所以只给出一个钩子：map_source，一个作用在挂载点正在构造的来源之上的变换。你的 crate 在它之上叠加自己的 trait，并约束到你自己的来源类型：

use ruststream::runtime::{Declared, SubscriberBuilder, SubscriberSettings};

pub trait NatsSubscriber {
    fn jetstream(self, stream: impl Into<String>) -> Self;
    fn durable(self, name: impl Into<String>) -> Self;
}

// 四个状态槽位依次是（工作者、失败策略、起始位置、批大小）；`Codec` 是这次注册自己的解码覆盖，
// 在没人指定之前是 `()`。两者都原样传递下去。
impl<Def, Workers, Failures, StartPosition, Batch, Codec> NatsSubscriber
    for SubscriberBuilder<Def, SubscribeOptions, (Workers, Failures, StartPosition, Batch), Codec>
where
    Def: Declared,
{
    fn jetstream(self, stream: impl Into<String>) -> Self {
        self.map_source(|source| source.jetstream(stream))
    }

    fn durable(self, name: impl Into<String>) -> Self {
        self.map_source(|source| source.durable(name))
    }
}

对来源类型的约束意味着，这些方法在别的 Broker 的构建器上根本不存在。下文 Out 槽位的词汇用的也是同一种扩展形态。

有一项核心设定改变的不是状态槽位，而是来源类型本身：start_at(..) 把描述符包进 StartAt<SubscribeOptions, Position>。于是恰恰在指定了起始位置的订阅上，你的方法不在作用域里，这种情形由第二个针对包装后来源的 impl 补上。StartAt::map_inner 取出里面的描述符，并原样交还位置，所以每个方法仍旧只有一行：

use ruststream::StartAt;
use ruststream::runtime::Fixed;

// 这里起始位置槽位按构造必然是 `Fixed` - 这个包装正是 `start_at(..)` 造出来的 - 而且源类型也
// 不同，所以这个 impl 与上面那个永远不会重叠。
impl<Def, Workers, Failures, Batch, Codec, Position> NatsSubscriber
    for SubscriberBuilder<
        Def,
        StartAt<SubscribeOptions, Position>,
        (Workers, Failures, Fixed, Batch),
        Codec,
    >
where
    Def: Declared,
{
    fn jetstream(self, stream: impl Into<String>) -> Self {
        self.map_source(|source| source.map_inner(|inner| inner.jetstream(stream)))
    }

    fn durable(self, name: impl Into<String>) -> Self {
        self.map_source(|source| source.map_inner(|inner| inner.durable(name)))
    }
}

用你自己的词汇表达发布者配置¶

发布这一侧是对称的。挂载点用 .out(marker, policy) 指定发布策略：标记 Reply 对应带 publish("dest") 的处理器返回的值，槽位的标记对应 Out 槽位。MapPublisher 就是作用在这个位置所持策略之上的钩子：

use ruststream::runtime::MapPublisher;

pub trait NatsPublish {
    fn stream(self, name: impl Into<String>) -> Self;
    fn expect_last_sequence(self, seq: u64) -> Self;
}

impl<T: MapPublisher<Policy = Publish>> NatsPublish for T {
    fn stream(self, name: impl Into<String>) -> Self {
        self.map_publisher(|policy| policy.stream(name))
    }

    fn expect_last_sequence(self, seq: u64) -> Self {
        self.map_publisher(|policy| policy.expect_last_sequence(seq))
    }
}

在服务里读起来是这样：

b.include(confirm).out_reply(Publish).stream("ORDERS");
b.include(mirror).out(Audit, Publish).stream("AUDIT").build();

约束落在策略上，而不是落在链上，所以一份实现同时适用于回复位置、每一个槽位、路由器和 Broker 作用域。

map_publisher 把策略换成同一类型的策略。换成另一种策略类型意味着另一种发布模式，它的位置在 .out(marker, policy) 调用本身。已经配置好的值也可以直接传到那里： .out_reply(Publish::default().stream("ORDERS"))。

发布构建器上的逐条设置¶

调用点通过你加到发布构建器上的一个步骤，去改你的 Publisher::Options 的某个字段。没有任何东西包裹发布者，所以这次发布仍然从挂载点自己的条目走出去，带着那个条目指定的编解码器和变换。

一共四块：一个每个字段都可选的设置类型，一份携带默认值的策略，一个把两者合起来的活发布者，以及一个以设置类型为约束、写在 PublishBuilder 上的扩展 trait。正是这个约束把你的步骤挡在别的 Broker 的发布者构建器之外：

tests/publish_options.rs

/// The broker's per-message settings. Every field optional: what a call leaves unset keeps what
/// the policy fixed. `Debug` and `PartialEq` are not part of the contract, they are what a test
/// naming this type needs to assert on it.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct PriorityOptions {
    priority: Option<u8>,
    ttl: Option<u8>,
}

/// The publish policy: pure declaration, constructible anywhere, and the place the defaults are
/// configured.
#[derive(Debug, Clone, Copy, Default)]
struct PriorityPublish {
    priority: u8,
    ttl: u8,
}

impl PriorityPublish {
    fn priority(mut self, priority: u8) -> Self {
        self.priority = priority;
        self
    }

    fn ttl(mut self, ttl: u8) -> Self {
        self.ttl = ttl;
        self
    }
}

/// The live publisher: the connection, plus the defaults the policy carried.
struct PriorityPublisher {
    inner: MemoryPublisher,
    default_priority: u8,
    default_ttl: u8,
}

impl PublishPolicy<ConnectedMemoryBroker> for PriorityPublish {
    type Live = PriorityPublisher;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(PriorityPublisher {
            inner: Publish.pair(connected).await?,
            default_priority: self.priority,
            default_ttl: self.ttl,
        })
    }
}

impl Publisher for PriorityPublisher {
    // Forwarded to the bus underneath, which keeps what it is handed.
    type Payload = Take;
    type Error = MemoryError;
    type Options = PriorityOptions;

    async fn publish(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let priority = options
            .and_then(|options| options.priority)
            .unwrap_or(self.default_priority);
        let ttl = options
            .and_then(|options| options.ttl)
            .unwrap_or(self.default_ttl);
        // A real broker hands the resolved values to its client as the protocol fields they are.
        // The in-memory bus has no such field, so this one puts them where a test can read them
        // back.
        let mut headers = msg.headers().clone();
        headers.insert("priority", priority.to_string());
        headers.insert("ttl", ttl.to_string());
        let stamped = OutgoingMessage::new(msg.name(), msg.payload()).with_headers(headers);
        self.inner.publish(stamped, None).await
    }
}

impl TransactionalPublisher for PriorityPublisher {
    async fn begin_transaction(&self) -> Result<(), Self::Error> {
        self.inner.begin_transaction().await
    }

    async fn commit(&self) -> Result<(), Self::Error> {
        self.inner.commit().await
    }

    async fn abort(&self) -> Result<(), Self::Error> {
        self.inner.abort().await
    }
}

/// The steps the broker puts in its prelude. The bound on the sink's options type is what keeps
/// them off a builder over any other broker's publisher.
trait PriorityPublishSteps {
    /// Sends this one message at `priority`, whatever the mount site's default is.
    #[must_use]
    fn priority(self, priority: u8) -> Self;

    /// Sends this one message with `ttl`, whatever the mount site's default is.
    #[must_use]
    fn ttl(self, ttl: u8) -> Self;
}

impl<Sink, Body, Enc, Hdrs, Dest> PriorityPublishSteps
    for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
where
    Sink: PublishSink<Options = PriorityOptions>,
{
    fn priority(mut self, priority: u8) -> Self {
        self.options_mut()
            .get_or_insert_with(PriorityOptions::default)
            .priority = Some(priority);
        self
    }

    fn ttl(mut self, ttl: u8) -> Self {
        self.options_mut()
            .get_or_insert_with(PriorityOptions::default)
            .ttl = Some(ttl);
        self
    }
}

宏路径和手写路径上，Broker 这一半是一样的。把扩展 trait 从你的 prelude 导出，就放在策略别名旁边。

逐条设置只有步骤这一种形状。不要把发送放进你自己的 trait：从你自己的值走出去的发布，槽位视图不再看得见，而排序键这类设置恰恰是测试要断言的东西。也不要用消息头携带它：这个设置是协议字段，而消息头只会把它变成一串字节，再让你的 publish 在同一个进程里解析回来。

你的 Broker 满足不了的值是一次发布错误，而不是悄悄退回默认值：调用方要的那个顺序，它拿不到。

能力 trait¶

只实现你的 Broker 真正支持的能力，它们都不属于必需接口。最接近必需的是 BatchSubscriber：能提供的地方就提供它，因为每个批量处理器都要它，而自身没有批量能力的传输照样可以在客户端攒批。

trait	适用于支持这些能力的 Broker
`BatchSubscriber`	按批接收消息
`TransactionalPublisher`	在句柄上围绕发布做 begin / commit / abort
`OwnedTransactions` / `Transaction`	同一个句柄上同时开启任意多个事务，每个各带自己的缓冲区
`RequestReply`	做原生的请求-响应
`Partitioned`	给出站消息设定分区键
`Seekable` / `Seeker`	在可重放的日志中重新定位一个活的订阅
`Positioned`	报告一次投递在日志中的位置
`DescribeServer`	为 AsyncAPI 报告一个 `ServerSpec`

Seekable 在 stream 借用订阅者之前交出 Seeker 句柄，因此可以从分发循环之外重新定位一个正在运行的订阅。

位置由 Broker 自己拥有：KafkaPosition 风格的构造函数由你在自己的类型上声明。通过 Positioned::position 从一条已投递消息上取到的位置确定了一条契约，定位到它会精确地重新投递这条消息。构造出来的位置，语义由你的位置类型自己写明。

写清楚一次定位的作用范围（一个消费者实例，还是一个共享的组游标），并重置这次定位所作废的一切 ack 记账。

要让处理器主体能够定位，就把投递位置和订阅的 seeker 放进投递上下文的字段，并为它们发布 ContextField 键。范本是内存 Broker 的 MemoryContext 及其 Position 和 SeekHandle 键。批量的那些写法从下面的批量上下文拿到 seeker，那里没有位置。

DescribeServer 给出的服务器描述，报告客户端所连接的主机和端口。凭据绝不出现在其中，因为这份文档就是为了发布而生成的。用 URL 配置的 Broker 通过 ServerSpec::from_url 构建描述，它会去掉 URL 里的用户名和密码。只去掉协议前缀、把剩下的部分照原样传进去，用户名和密码就留在了描述里：这正是 from_url 替掉的那个缺陷，它曾出现在不止一个 Broker crate 的发布版本里。配置了多个地址的 Broker，用 ServerSpec::host_from_url 把它们拼起来。

这些 trait 就是处理器主体所写的词汇。主体用它需要的那项能力约束自己的槽位（Out<impl TransactionalPublisher, Journal>，手动路径上是 where W: TransactionalPublisher），从不写你的任何类型。挂载点在编译期按这个约束检查一次所绑定策略的活形态。

四种发布者能力，每一种在 arena 条目上都有自己的类型化形态：发布构建器、事务作用域、拥有式事务、带关联的请求。这些形态建立在挂载点的编解码器和标记的字典之上。服务要用到它们，只需你在活的发布者上实现对应的 trait。

批：`BatchSubscriber`¶

接受 &[T] 的处理器消费的是一批消息，挂载点为它写下一个数字，也就是批大小。运行时把这个数字直接传给 BatchSubscriber::batches(size)。你的订阅者交出的批，就是处理器主体看到的批：运行时既不拆分也不合并，所以一批里绝不会超过 size 条消息，传输手上只有更少的消息时就更短。

把 size 对应到你的客户端已有的说法上：XREADGROUP COUNT、JetStream 的 pull 批、Kafka 的 poll 上限。至于一批怎么攒出来，其余的事（阻塞超时、消费者组、预取窗口）仍归你自己的词汇，通过你的设置扩展 trait 配置在订阅来源上。服务于是写成 b.include(handler.batch(nonzero!(6)).block(Duration::from_secs(5)))：核心的词在前，你的词在后。

把这项能力放到挂载能够到的每一个订阅者上，而不只是你自己的描述符打开的那一个。 #[subscriber("topic")] 走的是 Subscribe，所以这种写法下的 &[T] 主体，要的是 Subscribe::Subscriber 上的 BatchSubscriber。

只把能力挂在自家描述符订阅者上的 crate，会让字符串字面量那种写法编译不过。两者是同一个类型时无事可做，类型不同时两边都要有。

传输一次只投递一条消息时，也要实现这项能力，用核心的 BufferedSubscriber 在客户端攒批：它的 batches 遵守拿到的批大小。批大小不由你选，让不满的批提前结束的那个截止时间由你选，而且它不必是常量。

把这个截止时间放到你的订阅描述符上（.max_wait(Duration::from_millis(25))），在订阅打开时交给包装器，这样服务可以逐个订阅去调。10 ms 的默认值是照进程内总线定的：中间一旦隔着一次网络往返，大多数批会在一条投递上就结束，所以把这个截止时间做成描述符选项的 Broker crate 落在 10 到 50 ms 之间。

订阅者的其余部分原样穿过这层包装：

tests/batch_subscriber.rs

/// What a broker crate writes when its transport has no batches of its own: the subscriber it
/// already has, wrapped in the core's client-side buffer, and `BatchSubscriber` delegated to it.
/// The deadline that closes a partial batch is the broker's own choice; the batch size is not -
/// it arrives per subscription, as the argument of `batches`.
struct TrickleSubscriber(BufferedSubscriber<MemorySubscriber<Retaining>>);

impl TrickleSubscriber {
    fn new(inner: MemorySubscriber<Retaining>) -> Self {
        Self(BufferedSubscriber::new(inner).max_wait(Duration::from_millis(5)))
    }
}

impl Subscriber for TrickleSubscriber {
    type Message = <MemorySubscriber<Retaining> as Subscriber>::Message;
    type Error = <MemorySubscriber<Retaining> as Subscriber>::Error;

    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
        self.0.stream()
    }
}

impl BatchSubscriber for TrickleSubscriber {
    type Batch = Vec<<MemorySubscriber<Retaining> as Subscriber>::Message>;

    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, Self::Error>> + Send + '_ {
        self.0.batches(size)
    }
}

/// Buffering does not move the subscription, so every other capability reaches through the
/// wrapper unchanged - here the seeker, which is what lets a batch subscription open at a
/// position even where the batches are assembled on the client.
impl Seekable for TrickleSubscriber {
    type Seeker = <MemorySubscriber<Retaining> as Seekable>::Seeker;

    fn seeker(&self) -> Self::Seeker {
        self.0.seeker()
    }
}

/// The broker's own subscription descriptor, opening the batching subscriber above.
#[derive(Clone)]
struct Trickle {
    name: &'static str,
}

impl SubscriptionSource<ConnectedMemoryBroker<Retaining>> for Trickle {
    type Subscriber = TrickleSubscriber;
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        self.name
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker<Retaining>,
    ) -> Result<TrickleSubscriber, MemoryError> {
        Ok(TrickleSubscriber::new(
            Subscribe::subscribe(connected, self.name).await?,
        ))
    }
}

// A descriptor that addresses its own copies says where they go, and one in-memory subject is
// both ends of the bus, so the name answers for itself with no lookup in between.
impl RedeliveryAddressed<ConnectedMemoryBroker<Retaining>> for Trickle {
    fn redelivery_address(
        &self,
        _connected: &ConnectedMemoryBroker<Retaining>,
    ) -> impl Future<Output = Result<RedeliveryAddress, MemoryError>> + Send {
        ready(Ok(RedeliveryAddress::new(self.name)))
    }
}

挂载点看不出你走的是两条路里的哪一条：服务写下批大小，就拿到批。

在攒批会破坏传输自身某项保证的地方，不提供这项能力同样是正当的答复。ZeroMQ 的 ROUTER 是现实中的例子：它按每个对端各自的 reply-to 回复，而一整批只对应一个 PublishContext，于是这一批的回复都会发到某一个对端的地址上。在你的 crate 文档里写明这一点：&[T] 主体在那种传输上编译不过。

conformance 的批量套件会检查这项契约：它用小于一轮消息总数的批大小打开订阅，返回的批更长的 Broker 通不过。它不在 harness::run_suite 之内：能力套件由你自己调用，实现了哪项能力就调哪一个。

你的 crate 要提供的 prelude¶

你的类型由挂载点来写，而不是主体，这正是你的 crate prelude 的用处。提供一个 prelude 模块，按这个顺序分三层：

pub use ruststream::prelude::*;，让一个 glob 就能服务整个文件；
服务会写到的、你自己的那部分表面：Broker、它的订阅来源、它的 Config、它的错误，以及主体会读的 ContextField 键；
你的发布策略，用每个 Broker 都用的那套统一名字：Publish，以及在你有的时候还有 TransactionalPublish 和 Request（pub use crate::KafkaTransactionalPublish as TransactionalPublish;）。再把你在活值上实现的能力 trait 作为一份清单加进去，这样带来策略的那个 glob 也会把它们的操作带进作用域。

核心 prelude 在这三个名字下什么都不导出，所以挂载点在哪个 Broker 上读起来都一样。切勿把策略取名成核心 trait 的名字（Publisher、TransactionalPublisher、OwnedTransactions、RequestReply），也不要在这些名字下重导出别的东西：同时 glob 了两个 prelude 的主体，必须仍然把这些名字解析成核心 trait。

清单指的是你的 glob 添加的那部分，也就是主体经由你的 Broker 才够得到的消费侧 trait：Positioned、 Seeker、Transaction 之类。四个发布者能力已经在核心 prelude 里，再导出一遍什么也不改变。

某个 trait 的方法会和核心的默认方法冲突时，就把它留在清单外面（实践中就是 Partitioned::partition_key 与 IncomingMessage::partition_key），让需要它的服务显式导入。 BatchSubscriber 不属于任何清单：调用它的是框架，没有哪个主体会把它写成约束。

可以参照的现成例子是 ruststream::memory::prelude。

扩展 `Out` 槽位的词汇¶

处理器参数 Out<impl X, Marker> 接受槽位背后那个活值实现了的任意 X。在此之上，核心还会转发它自己的那套能力（Publisher、TransactionalPublisher、OwnedTransactions、RequestReply）。活值提供的能力不止于此，或者它根本就不是发布者（一个按分区的 producer 缓存、一个分片路由器）时，就声明你自己的能力 trait，并为这个活值实现它。

处理器主体手里拿到的不是那个值，而是 arena 里的条目 Slot<Marker, W, E, Pipe, Body>，一扇通向它的透明窗口。自动解引用能把方法调用送过这扇窗，却送不过 trait 约束：写成 fn issue<L: Lanes>(lanes: &L) 的辅助函数会以 E0277 拒收这个条目。

在你的 trait 旁边加上一个 blanket 实现 impl<M, W: Lanes, E, Pipe, Body> Lanes for Slot<M, W, E, Pipe, Body>，通过条目的 Deref 转发，按能力泛型的辅助函数和主体就能原样接收这个条目。具体类型依然不会出现在应用代码里：

宏手写

tests/out_slots.rs

// A paired value that is NOT a publisher: a lane router in the shape of a broker's
// per-partition producer cache. The capability is broker-defined; the core knows nothing
// about it.
#[derive(Clone)]
struct LaneRouter {
    publisher: MemoryPublisher,
}

/// The broker-defined capability: pick a destination lane for a shard.
trait ShardLanes {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str);
}

impl ShardLanes for LaneRouter {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        let dest = if shard.is_multiple_of(2) {
            "slots.lane.even"
        } else {
            "slots.lane.odd"
        };
        (&self.publisher, dest)
    }
}

// Grafted onto the arena entry once, for every marker, delegating through the entry's
// transparent `Deref`: this is how a broker crate extends the slot vocabulary with its own
// traits. A handler body holds the entry, so without this impl the capability is reachable by
// autoderef for a method call but never satisfies a trait bound.
impl<M, W: ShardLanes, E, Pipe, Body> ShardLanes for Slot<M, W, E, Pipe, Body> {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        (**self).lane(shard)
    }
}

/// The bound the graft buys: a helper generic over the capability, not over the concrete live
/// type, takes the entry a handler body holds.
async fn sent<L: ShardLanes + Sync>(lanes: &L, event: &Event) -> bool {
    let (publisher, dest) = lanes.lane(event.id);
    publisher.message(event).to(dest).publish().await.is_ok()
}

/// The policy half: pure declaration pairing into the router, like a broker's
/// `per_partition()` policy pairs into its producer cache. No `Clone`: resolution consumes it.
struct LanePolicy;

impl PublishPolicy<ConnectedMemoryBroker> for LanePolicy {
    type Live = LaneRouter;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(LaneRouter {
            publisher: Publish.pair(connected).await?,
        })
    }
}

/// The handler bounds its slot with the broker-defined capability, not a core one.
#[subscriber("slots.sharded")]
async fn route_shard(event: &Event, Out(lanes): Out<impl ShardLanes>) -> HandlerOutcome {
    if sent(lanes, event).await {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

tests/manual_out_slots.rs

// A paired value that is NOT a publisher: a lane router in the shape of a broker's
// per-partition producer cache. The capability is broker-defined; the core knows nothing
// about it.
#[derive(Clone)]
struct LaneRouter {
    publisher: MemoryPublisher,
}

/// The broker-defined capability: pick a destination lane for a shard.
trait ShardLanes {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str);
}

impl ShardLanes for LaneRouter {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        let dest = if shard.is_multiple_of(2) {
            "slots.lane.even"
        } else {
            "slots.lane.odd"
        };
        (&self.publisher, dest)
    }
}

// Grafted onto the arena entry once, for every marker, delegating through the entry's
// transparent `Deref`: this is how a broker crate extends the slot vocabulary with its own
// traits. A body holds the entry, so without this impl the capability is reachable by autoderef
// for a method call but never satisfies a trait bound.
impl<M, W: ShardLanes, E, Pipe, Body> ShardLanes for Slot<M, W, E, Pipe, Body> {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        (**self).lane(shard)
    }
}

/// The bound the graft buys: a helper generic over the capability, not over the concrete live
/// type, takes the entry a body holds.
async fn sent<L: ShardLanes + Sync>(lanes: &L, event: &Event) -> bool {
    let (publisher, dest) = lanes.lane(event.id);
    publisher.message(event).to(dest).publish().await.is_ok()
}

/// The policy half: pure declaration pairing into the router, like a broker's
/// `per_partition()` policy pairs into its producer cache. No `Clone`: resolution consumes it.
struct LanePolicy;

impl PublishPolicy<ConnectedMemoryBroker> for LanePolicy {
    type Live = LaneRouter;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(LaneRouter {
            publisher: Publish.pair(connected).await?,
        })
    }
}

/// The body leaves the wired live value generic and bounds it with the broker-defined
/// capability, exactly as the attribute's `Out<impl ShardLanes>` does.
struct RouteShard;

struct Lanes;

impl OutSlot for Lanes {
    const NAME: &'static str = "Lanes";
    type Destination = Reads;
}

impl<L> Handle<Event, (), Outs<(L,)>> for RouteShard
where
    L: OutEntry<Lanes, Wire: ShardLanes>,
{
    async fn handle(
        &self,
        event: &Event,
        outs: &Outs<(L,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if sent(outs.get(Lanes), event).await {
            Ok(())
        } else {
            Err(HandlerOutcome::retry())
        }
    }
}

决定 trait 形状的是发送发生在哪里，而形状有两种。

路由器形状的能力交出一个发布者，自己从不发送：上面那个按分区的 producer 缓存为某个分片挑出发布者并把它返回。经这个发布者做的发布走在槽位视图之外，所以测试套件不会把它记到槽位名下，就像一个已结算的 owned 事务的缓冲区那样。改在 Broker 的发布日志上断言它。这是归属的边界，也是交出内层发布者所付的代价。

步骤形状的能力设定一项逐条设置，并以一次发布收尾：一个排序键、一个优先级、一个 QoS。这一种根本不是能力 trait：它是你的 Publisher::Options 的一个字段，由发布构建器上的一个步骤去改。参见发布构建器上的逐条设置。

你这个 crate 的 prelude¶

两种文件导入的东西不一样，正是这种分工让服务保持可移植。处理器主体导入 ruststream::prelude::*，不导入你的任何东西：它用核心的能力 trait 去约束注入进来的槽位，也就是 Out<impl Publisher>、 Out<impl TransactionalPublisher>、Out<impl OwnedTransactions>、Out<impl RequestReply>，于是主体说清楚它对发布者有什么要求，却从不说这是哪个 Broker 提供的。

挂载文件导入你的 prelude，因为指名 Broker 的地方就在那里。

只有一个例外：逐条设置，它的调用点在主体里。改动它的主体为了那个步骤导入你的 prelude，并在约束里写出你的设置类型（Out<impl Publisher<Options = MqttOptions>, Telemetry>）。这样的主体绑在你的 Broker 上，它的签名也把这一点说了出来。

这样一来，你的 prelude 就是使用你这个 Broker 的服务所写的那一个导入，它的形状因此属于契约的一部分。策略别名（NatsPublish as Publish、KafkaTransactionalPublish as TransactionalPublish、 LapinRequest as Request）让挂载文件在哪个 Broker 上读起来都一样，换 Broker 就是换一行导入。

你这一半的命名规则是：显式 re-export 会一声不响地盖过 glob，所以一个跟核心 trait 同名的名字，会把那个 trait 从每个写了这行 glob 的服务手里拿走，而错误出现在服务的文件里，不在你的文件里。

用一个跟在自己 glob 后面的探针把两半都固定下来：主体所写的那个约束仍然必须解析成核心 trait，挂载点的名字仍然必须是你的策略。

// in your crate, behind your own prelude glob
use crate::prelude::*;

// A capability bound a body states: the core trait, not something of yours.
fn _p<T: Publisher>() {}

// A mount-site name: your policy, constructible with no connection in sight.
fn _q() {
    let _: Publish = Publish::default();
}

单条投递的上下文与 `Ctx` 键¶

Broker 有原生的投递元数据（一个分区、一个偏移量、一个流序号）时，把它作为类型化的单条投递上下文暴露出来：一个由订阅者指明的 #[non_exhaustive] 结构体，外加若干 ContextField 键类型。处理器按键用 Ctx<K> 提取器把单个字段绑定成参数。键是空结构体，投递路径上既没有 type-map，也没有堆分配。

/// Per-delivery context of this broker.
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct MyContext {
    pub partition: i32,
}

/// `Ctx<Partition>` in a handler binds the delivery's partition.
#[derive(Debug, Default, Clone, Copy)]
pub struct Partition;

impl ContextField for Partition {
    type Context = MyContext;
    type Value = i32;
    fn read(self, src: &MyContext) -> i32 {
        src.partition
    }
}

这段草图读的是一个 Copy 标量，拥有和借用没有分别。位置类型不是 Copy 时（Pulsar 的消息 id、 Kinesis 的分片加序列号字符串），就以借用的方式读：Field::Value<'a> 对来源的生命周期是泛型的，所以键交回的是 &'a MessageId，用 ctx.context(..) 读它的主体一份都不必复制。

必须拥有所有权且为 'static 的只有 ContextField::Value，也就是 Ctx<K> 提取器背后的那个值，因为提取器的值在主体运行之前就要绑定好；这个键把借用形态交回的东西克隆一份。一个键通常两个 trait 都实现，各出一种形状。

没有单条投递字段的 Broker 用 ()。

批量订阅另有自己的上下文，因为一批横跨多次投递。把整条订阅共享的东西（seek 句柄、流的名字、消费者组）攒成第二个结构体，在它上面实现 BuildBatchContext，再发布若干 Field 键，好让批量主体用 ctx.context(..) 读它。运行时按批构造一个值，取自这一批的第一次投递。

逐次投递的字段不放进去：位置属于某一次投递，所以由批从元素上读。把两个结构体分开，正是让这条规则在编译期成立的办法：投递上下文不实现 BuildBatchContext，批量主体也就写不出它。

范本是内存 Broker 的 MemoryBatchContext：订阅的 seeker 用的还是它的投递上下文所发布的那个 SeekHandle 键。订阅这一级上没有东西可交的 Broker 什么都不用实现，批量停在 () 这个默认值上。

异步边界上的中间件¶

需要围绕编码和解码做异步 I/O 的集成（一个 schema 注册表、一层传输格式的信封）不属于 Codec：核心的编解码器是同步的，处理器也应当继续用默认的那一个。

把这类集成放到异步边界上。入站载荷在订阅的投递路径上转码，在编解码器看到它们之前完成。出站的用核心的 PublishLayer 加上信封，通过 RustStream::publish_layer 添加到整个应用上。发布层是异步的，也可以返回错误，而 Outgoing::payload_mut 的存在正是为了包装信封。

协议绑定¶

生成的 AsyncAPI 文档留出了只有你的 Broker 才知道的那部分：RabbitMQ 队列是否持久、Kafka 的消费者组、 MQTT 的 QoS。规范把这些叫做绑定，填写它们的是你的描述符。

tests/asyncapi.rs

/// A descriptor of the shape a broker crate ships: it reads its own private fields and says
/// what the protocol calls them.
#[derive(Clone)]
struct RabbitQueue {
    name: &'static str,
    durable: bool,
}

impl<C: Subscribe> SubscriptionSource<C> for RabbitQueue {
    type Subscriber = C::Subscriber;
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        self.name
    }

    async fn subscribe(self, connected: &C) -> Result<Self::Subscriber, C::Error> {
        connected.subscribe(self.name).await
    }

    fn channel_bindings(&self) -> Bindings {
        let body = AmqpChannel {
            is: "queue",
            queue: AmqpQueue {
                name: self.name,
                durable: self.durable,
            },
        };
        Binding::new("amqp", "0.3.0", &body)
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }

    fn operation_bindings(&self) -> Bindings {
        Binding::new("amqp", "0.3.0", &AmqpOperation { ack: true })
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }

    fn message_bindings(&self) -> Bindings {
        let body = AmqpMessage {
            message_type: "order",
        };
        Binding::new("amqp", "0.3.0", &body)
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }
}

Binding::new(protocol, version, &body) 把 body 序列化一次，并且自己写上 bindingVersion，所以你不会发布一个没有版本的绑定。协议键由内核对照规范的封闭列表检查，未列出的键会返回错误，而不是进入一份没有工具读得懂的文档。Bindings 默认为空：什么也不说的描述符不会改变任何文档。

服务器这一层是字段而不是方法，因为一个 Broker 只描述一次服务器：在你的 DescribeServer 实现里写 ServerSpec::new(host, protocol).bindings(..)。

绑定里应该放什么，由三条规则限定。

值只由描述符本身算出。文档在连接之前就生成，所以 Kafka 主题真实的分区数、Pub/Sub 订阅背后的主题、 SQS 队列的 ARN，都不可能从这里报出来。

凭据永远不进绑定，理由和 DescribeServer 一样。检查放在 conformance::harness 里：用一个已知的密码配置你的 Broker 和描述符，然后跑这个扫描。

tests/conformance_self.rs

/// The in-memory broker has no network address and no binding, so nothing it describes can leak a
/// password. A broker configured from a URL runs this with the password it was configured with.
#[cfg(feature = "asyncapi")]
#[test]
fn memory_broker_describes_without_credentials() {
    harness::describes_without_credentials(
        &MemoryBroker::new(),
        &MemorySource::new("orders"),
        "hunter2",
    );
}

规范没有为之提供绑定的协议，走 Binding::extension("x-kinesis", &body)。协议键是一个封闭列表，所以 ZeroMQ、Kinesis 和文件传输都没有合法的键；x- 扩展位于同一层级，并且不带 bindingVersion。

绑定来自描述符，所以按裸名字打开的订阅一个也没有：那里没有什么可描述的。想要绑定的 Broker，要提供一个 SubscriptionSource 类型。

另一侧的同样三个名字，由你的发布策略填写。回复、Out 槽位，以及死信投递所经过的发布者，都是 PublishPolicy，各自描述自己发布到的通道。

tests/asyncapi.rs

/// A publish policy of the shape a broker crate ships: an SNS topic is named by its binding's
/// required `name`, and the hook is handed the destination the mount site resolved.
#[derive(Clone, Copy, Default)]
struct TopicPublish;

impl PublishPolicy<ConnectedMemoryBroker> for TopicPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        live(connected)
    }

    fn channel_bindings(&self, channel: &str) -> Bindings {
        let body = SnsChannel {
            name: channel.to_owned(),
        };
        one("sns", "0.1.0", &body)
    }

    fn operation_bindings(&self, channel: &str) -> Bindings {
        let body = SnsOperation {
            topic: SnsChannel {
                name: channel.to_owned(),
            },
        };
        one("sns", "0.1.0", &body)
    }

    fn message_bindings(&self, _channel: &str) -> Bindings {
        let body = SnsMessage {
            message_type: "progress",
        };
        one("sns", "0.1.0", &body)
    }
}

每个方法拿到的，是挂载点解析出来的目的地。回复拿到的是回复类型自己的名字，或者注册上的 publish("dest") 子句；槽位条目拿到的是它自己的名字；死信投递拿到的是 dead_letter("dlq") 声明。 SNS 主题和 SQS 队列由各自绑定里必填的 name 指名，这个名字就从这里来：策略持有的是你的 Broker 的设置，从来不是目的地。转换逐条投递指定目的地时，通道报不出地址，方法拿到的是挂载点的后备名字。

描述符这一侧没有这个参数：它知道自己描述的是哪个订阅。

那三条规则在这里原样成立。有一点不适用：回复没有自己的 send 操作，所以回复策略上的 operation_bindings 到不了文档。槽位和死信目的地各有一个。

第四个方法只属于回复。如果你的 Broker 通过 reply-to 头路由回复，reply_address_location 就说明客户端从哪里读这个地址：

tests/asyncapi.rs

/// The reply's own policy: it answers where a client reads the address of an answer.
#[derive(Clone, Copy, Default)]
struct ReplyToPublish;

impl PublishPolicy<ConnectedMemoryBroker> for ReplyToPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        live(connected)
    }

    fn channel_bindings(&self, channel: &str) -> Bindings {
        let body = NatsChannel {
            subject: channel.to_owned(),
        };
        one("nats", "0.1.0", &body)
    }

    fn reply_address_location(&self) -> Option<&'static str> {
        Some("$message.header#/reply-to")
    }
}

这时文档把回复通道报成 address: null，并把该表达式放进 receive 操作的 reply.address.location。只有挂载点组合了逐条投递指定目的地的转换时才会用到它；否则回复去往声明的名字，文档报出的也是这个名字。

规范未列出的协议，在发布这一侧同样没有绑定可填。内存 Broker 就是这种情况：不存在 memory 这个键，所以 MemoryPublish 保持沉默，而不是自己造一个。

这些钩子由内核的 asyncapi feature 控制。从你的 crate 里转发它：

[features]
asyncapi = ["ruststream/asyncapi"]

并在你填写的每个方法上加 #[cfg(feature = "asyncapi")]。

配置与默认值¶

Config 归你的 crate 所有，核心不带任何 Broker 专有的配置。某个字段没有合理的默认值时，就不要实现 Default：用户于是显式写下这个值，而不是继承一个日后会出问题的默认值。

错误¶

用 thiserror 写一个 crate 级别的错误枚举，变体按来源划分。公开的错误枚举标记 #[non_exhaustive]。切勿在库 crate 里使用 anyhow。

测试支持¶

在 testing feature 下提供一个进程内传输，在它的已连接形态上实现 TestableBroker。用 register_testable_broker! 为这个已连接类型注册：测试套件会先连接每一个 Broker，然后才取回它的传输。用户于是可以借助 TestApp，对着你的 Broker 单元测试处理器。

该传输只做核心路由：把发布出去的消息分发给匹配的订阅者，对 ack 和 nack 的答复与真实传输一致。传输能确认的地方，就在内存里结算，nack(requeue = true) 把这条投递放回去。传输根本无法确认的地方（ZeroMQ、MQTT QoS 0、Redis pub/sub），答复仍然是 AckError::Unsupported。它一旦声称一次真实传输做不到的结算，处理器里的重试就会在测试里通过，在生产中丢消息。

切勿在传输里模拟 Broker 专有的语义（持久游标、重新投递定时器、偏移量、死信路由），那些要对着一台真实的服务器端到端地验证。

参考实现就是内存 Broker 自己的那一份（在 ConnectedMemoryBroker 上）：

src/memory/mod.rs

// The harness drives the connected form: TestApp connects every registered broker before it
// recovers the in-process transport, and run_suite scenarios receive connected brokers.
#[cfg(feature = "testing")]
impl<Log: LogMode> crate::testing::TestableBroker for ConnectedMemoryBroker<Log> {
    fn install_coordinator(&self, coordinator: Coordinator) {
        self.state.install_coordinator(coordinator);
    }

    fn inject(&self, message: OutgoingMessage<'_>) {
        let (name, payload, headers) = message.into_parts();
        // Injecting into a shut-down bus is a harness bug (both run_suite and TestApp drive
        // the bus strictly before shutdown), so fail loudly instead of losing the message.
        self.state
            .fanout(name, Bytes::copy_from_slice(payload), headers)
            .expect("inject on a shut-down broker: drive the harness before shutdown");
    }

    /// What the broker holds under `name`: everything published there on a discarding broker
    /// (which records for the length of a harness run), and the retained window on a retaining
    /// one, so an assertion never claims more than the broker keeps.
    fn published(&self, name: &str) -> Vec<RawMessage> {
        self.state
            .log
            .lock()
            .expect("memory broker mutex poisoned")
            .name(name)
            .map(|log| log.messages(name))
            .unwrap_or_default()
    }
}

// One registration per log mode: the harness recovers a broker by its concrete type, and the
// two modes are two types.
#[cfg(feature = "testing")]
crate::register_testable_broker!(ConnectedMemoryBroker<Discarding>);
#[cfg(feature = "testing")]
crate::register_testable_broker!(ConnectedMemoryBroker<Retaining>);

该传输在每次把消息入队给某个订阅者时调用 Coordinator::enqueued，在结算或丢弃一次投递时调用 Coordinator::consumed，测试套件据此判断这次反应已经结束。延迟的重新投递由它交给 Coordinator::schedule_redelivery 去路由。

同一个类型既适用于 TestApp，也适用于 conformance 校验套件。面向用户的那一侧参见测试；Conformance 讲的是怎样用 run_suite 和 lifecycle 转移链检查证明你的实现。

怎样写一个信得过的进程内传输¶

一个服务的整套测试都跑在这个进程内传输上。因此它和真实传输之间的每一处差异，都会让一个测试变绿，而它测的行为在生产里并不存在。这些差异并不冷僻，而下面每一条规则的代价大约就是一个测试。

核心的契约套件不能只跑真实服务器，也要跑进程内传输。套件是照着 trait 写的，并不区分应答的是真实 Broker 还是进程内传输。一个 #[tokio::test] 就够：

tests/conformance_self.rs

/// The suites that read a broker's publish log back - the routing contract's log check and the
/// seeking capability - need a broker that keeps one.
fn replaying() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(64)))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_conformance_suite() {
    harness::run_suite(replaying).await;
}

先跑 lifecycle。它会走一遍 new -> connect -> 订阅 -> 发布 -> ack -> shutdown，然后问出一个几乎没人拿去问进程内传输的问题：关闭之前创建的发布者，在关闭之后会不会返回错误？真实客户端答的是 “未连接”。而发布只是往 channel 里发一条消息的进程内传输没有理由返回错误，它会把消息收下。

tests/conformance_self.rs

// `make_source` / `make_publisher` must stay closures: their bounds are higher-ranked
// (`Fn(&str) -> _` / `Fn(&B) -> _`), so a bare method path - which binds one concrete lifetime -
// would not type-check.
#[allow(clippy::redundant_closure, clippy::redundant_closure_for_method_calls)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_lifecycle() {
    harness::lifecycle(
        MemoryBroker::new,
        |name| MemorySource::new(name),
        |broker| broker.publisher(),
    )
    .await;
}

capabilities::* 套件照此加上，实现了哪项能力就加哪一个。

提供的能力要和真实 Broker 一致。testing 这个 feature 是给测试用的，release 构建会把它关掉，两个方向的代价因此并不对等。少一项是贵的：真实 Broker 有、进程内传输没有的能力，在进程内根本挂载不了，它背后的行为也就没人测。多一项是便宜的：只有进程内传输提供的事务或 request-reply，在你自己的 release 构建里就编译不过，恼人，但立刻就能发现。

传输怎么结算，你就怎么结算。真实的 ack 在两处返回 AckError::Unsupported：发完即忘的传输，以及至多一次的服务质量。进程内传输照样返回它。为了让套件通过而回一个 Ok(())，会让一个返回 HandlerOutcome::retry() 的处理器在进程内通过测试，却在真实 Broker 上丢消息。诚实的答复，套件是接受的。

客户端做的事要复刻，Broker 做的事不要假造。这条界线无关工作量，只看机制运行在哪一侧。竞争消费、按组分发、关联与回复路由、提交前的缓冲，都由客户端或路由层完成，进程内复刻是精确的。集群原子性、fencing、Broker 侧持有的超时和 exactly-once 由 Broker 完成，进程内复刻就是虚构。

竞争消费最该做对，因为做错了看着像成功。把每条消息都发给队列的每个订阅者，那已经不是队列。共用一条队列的两个工作者于是各自跑完整个流，而一个统计处理条数的测试只看到消息都处理完了，一个错都不报。

每一处缺口都配一句注释，点名它让哪条断言站不住。不要写“这个功能没有”，要写读者从此不该再信哪个测试，以及真正验证它的是什么：

// No fencing: a second producer claiming the same transactional id is not rejected here, so a
// test cannot assert the first one is fenced out. `capabilities::transactions` against a real
// server is what covers that.

缺口还要用一个测试守住。某人“修好”进程内传输，让它去路由那些它故意不路由的东西，注释就在那天失效。而一个断言这个处理器没有运行的测试，会在那天失败，并且自己把话说清楚。

挂载进程内传输的方式，要和挂载真实 Broker 一样。你自己的订阅来源和发布策略必须原封不动地对着它工作，这样服务测的才是它实际交付的那份路由文件。用户非得把 OrdersStream 换成别的东西才能把测试跑起来，这个测试就不再检验挂载了。

编写一个 Broker¶

必需的 trait¶

Broker 与 ConnectedBroker¶

Subscribe¶

Subscriber¶

IncomingMessage¶

Publisher¶

PublishPolicy¶

订阅来源¶

谁来发布重试副本¶

注册声明了什么¶

重试副本发往哪里¶

用一个字符串命名一种订阅方式¶

用你自己的词汇表达配置¶

用你自己的词汇表达发布者配置¶

发布构建器上的逐条设置¶

能力 trait¶

批：BatchSubscriber¶

你的 crate 要提供的 prelude¶

扩展 Out 槽位的词汇¶

你这个 crate 的 prelude¶

单条投递的上下文与 Ctx 键¶

异步边界上的中间件¶

协议绑定¶

配置与默认值¶

错误¶

测试支持¶

怎样写一个信得过的进程内传输¶

`Broker` 与 `ConnectedBroker`¶

`Subscribe`¶

`Subscriber`¶

`IncomingMessage`¶

`Publisher`¶

`PublishPolicy`¶

批：`BatchSubscriber`¶

扩展 `Out` 槽位的词汇¶

单条投递的上下文与 `Ctx` 键¶