ruststream 0.7.5

//! A broker's per-message settings: defaulted by the publish policy at the mount site, adjusted
//! per call by steps the broker adds to the publish builder.
//!
//! The broker half here is what a broker crate ships - an options type whose every field is
//! optional, a policy carrying the defaults, a live publisher resolving one against the other,
//! and an extension trait bounded on the options type so its steps appear on a builder over this
//! publisher and nowhere else. The in-memory bus has no protocol field to put a priority in, so
//! the publisher folds the resolved value into a header the assertions read back.
#![cfg(all(
    feature = "memory",
    feature = "macros",
    feature = "json",
    feature = "cbor",
    feature = "testing"
))]

mod common;

use common::{Order, Receipt};

use ruststream::codec::CborCodec;
use ruststream::memory::prelude::*;
use ruststream::memory::{ConnectedMemoryBroker, MemoryError, MemoryPublisher};
use ruststream::runtime::{
    ContextKind, ForSlot, Outgoing, PublishBuilder, PublishSink, PublishTransform, Reads,
    SlotContext, for_batch,
};
use ruststream::testing::TestApp;
use ruststream::{BytesMut, OutgoingMessage, PairError, Take};

/// 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
    }
}

#[derive(OutSlot)]
#[publishes(Order)]
struct Ledger;

// --8<-- [start:handler]
// A body that adjusts a per-message setting names the broker's step, so it imports that broker's
// prelude - the stated exception to "a body imports the framework prelude alone".
#[subscriber("options.in")]
async fn record(
    order: &Order,
    Out(ledger): Out<impl Publisher<Options = PriorityOptions>, Ledger>,
) -> HandlerOutcome {
    if ledger
        .message(order)
        .to("options.normal")
        .publish()
        .await
        .is_err()
        || ledger
            .message(order)
            .to("options.urgent")
            .priority(9)
            .publish()
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}
// --8<-- [end:handler]

/// What a call leaves alone is what the mount site fixed, and a step wins over it for that one
/// message. Both publishes leave the same slot, in the same handler.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_mount_site_default_holds_until_a_step_overrides_it() {
    // --8<-- [start:mount]
    let app =
        RustStream::new(AppInfo::new("options", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3))
                .build();
        });
    // --8<-- [end:mount]
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Order>("options.normal")
        .assert_called_once()
        .with(&Order { id: 7 })
        .with_header("priority", "3");
    tb.broker::<MemoryBroker>()
        .published::<Order>("options.urgent")
        .assert_called_once()
        .with(&Order { id: 7 })
        .with_header("priority", "9");
}

/// The defect this design closes: a step is a position on the builder, not a wrapper around the
/// publisher, so the publish it finishes still encodes with the codec the mount site named.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_step_keeps_the_codec_the_mount_site_named() {
    let app = RustStream::new(AppInfo::new("options-codec", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3))
                .codec(CborCodec)
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");

    // Both messages are CBOR, the slot's codec - the stepped one included.
    tb.out::<Ledger>()
        .assert_called(2)
        .decoded_as::<Order>()
        .with_codec(&CborCodec, &Order { id: 7 });
    tb.broker::<MemoryBroker>()
        .published::<Order>("options.urgent")
        .assert_called_once()
        .with_codec(&CborCodec, &Order { id: 7 })
        .with_header("priority", "9");
}

#[derive(OutSlot)]
#[publishes(Order)]
struct Staged;

/// A transaction opened on the slot publishes through the same builder, so the same step reaches
/// the buffered message.
#[subscriber("options.txn.in")]
async fn stage(
    order: &Order,
    Out(staged): Out<impl TransactionalPublisher<Options = PriorityOptions>, Staged>,
) -> HandlerOutcome {
    let Ok(scope) = staged.begin().await else {
        return HandlerOutcome::retry();
    };
    if scope
        .message(order)
        .to("options.staged")
        .priority(5)
        .ttl(6)
        .publish()
        .await
        .is_err()
    {
        let _ = scope.abort().await;
        return HandlerOutcome::retry();
    }
    if scope.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_step_inside_a_transaction_scope_reaches_the_broker() {
    let app = RustStream::new(AppInfo::new("options-txn", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(stage)
                .out(Staged, PriorityPublish::default().priority(3))
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 4 })
        .to("options.txn.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Order>("options.staged")
        .assert_called_once()
        .with(&Order { id: 4 })
        .with_header("priority", "5")
        .with_header("ttl", "6");
    // A transaction scope publishes through the slot's own publisher, so the slot view records
    // the step the buffered message carried.
    tb.out::<Staged>()
        .assert_called_once()
        .with_options(&PriorityOptions {
            priority: Some(5),
            ttl: Some(6),
        });
}

/// The steps are on the builder, so they are there on a bare publisher too - here the one a
/// startup hook is handed.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_bare_publisher_takes_the_same_steps() {
    let app = RustStream::new(AppInfo::new("options-startup", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.after_startup(
                PriorityPublish::default().priority(3),
                async move |publisher: PriorityPublisher| {
                    publisher
                        .message(&Order { id: 1 })
                        .to("options.announce")
                        .publish()
                        .await?;
                    publisher
                        .message(&Order { id: 2 })
                        .to("options.announce.urgent")
                        .priority(8)
                        .publish()
                        .await
                },
            );
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");
    tb.settle().await.expect("settle");

    tb.broker::<MemoryBroker>()
        .published::<Order>("options.announce")
        .assert_called_once()
        .with_header("priority", "3");
    tb.broker::<MemoryBroker>()
        .published::<Order>("options.announce.urgent")
        .assert_called_once()
        .with_header("priority", "8");
}

/// A reply has no call site: the policy the mount site names for the `Reply` position holds the
/// defaults, and only a transform adjusts them.
#[subscriber("options.reply.in", reply("options.reply.out"))]
async fn acknowledge(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

#[derive(OutSlot)]
#[publishes(Order)]
struct Notes;

/// A body that names no step: the publish carries no options, and the policy's own setting is the
/// whole answer.
#[subscriber("options.note.in")]
async fn note(
    order: &Order,
    Out(notes): Out<impl Publisher<Options = PriorityOptions>, Notes>,
) -> HandlerOutcome {
    if notes
        .message(order)
        .to("options.note.out")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The broker folds the options into its protocol and the publish log sees only the result, so
/// the slot view is where a test reads back what the call site asked for.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_slot_view_reads_back_the_options_a_publish_carried() {
    let app = RustStream::new(AppInfo::new("options-view", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3))
                .build();
            b.include(note)
                .out(Notes, PriorityPublish::default().priority(3))
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");
    tb.message(&Order { id: 7 })
        .to("options.note.in")
        .publish()
        .await
        .expect("publish");

    // --8<-- [start:options_assert]
    tb.out::<Ledger>()
        .assert_called(2)
        .with_options(&PriorityOptions {
            priority: Some(9),
            ttl: None,
        });
    tb.out::<Notes>()
        .assert_called_once()
        .assert_options_default();
    // --8<-- [end:options_assert]
}

/// One delivery through `record`: the slot view holds an unstepped publish followed by a stepped
/// one, and the broker's log holds both.
async fn one_delivery() -> TestApp<()> {
    let app = RustStream::new(AppInfo::new("options-panics", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3))
                .build();
            b.include(note)
                .out(Notes, PriorityPublish::default().priority(3))
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");
    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");
    tb.message(&Order { id: 7 })
        .to("options.note.in")
        .publish()
        .await
        .expect("publish");
    tb
}

/// Another broker's options type, for the assertion that names the wrong one.
#[derive(Debug, PartialEq, Eq)]
struct TtlOptions {
    ttl: Option<u8>,
}

/// The broker's view reads the options of a slot publish too: the harness records them where the
/// runtime hands the message over, before the broker resolves them into its protocol.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_broker_view_reads_the_options_of_a_slot_publish() {
    let tb = one_delivery().await;

    tb.broker::<MemoryBroker>()
        .published::<Order>("options.urgent")
        .assert_called_once()
        .with_options(&PriorityOptions {
            priority: Some(9),
            ttl: None,
        });
}

/// Naming another broker's options type is the same mistake, and the panic names the type that
/// was actually recorded.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "PriorityOptions`, not `")]
async fn options_of_another_type_name_the_recorded_one() {
    let tb = one_delivery().await;

    tb.out::<Ledger>()
        .assert_called(2)
        .with_options(&TtlOptions { ttl: Some(9) });
}

/// Expecting a value where no step ran reads as a missing step, so the panic says the publish
/// took the policy's defaults.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "published with the policy's default options")]
async fn expecting_options_on_an_unstepped_publish_says_so() {
    let tb = one_delivery().await;

    tb.out::<Notes>()
        .assert_called_once()
        .with_options(&PriorityOptions {
            priority: Some(9),
            ttl: None,
        });
}

/// The mirror mistake: a step did run, so the assertion that nothing was set names what it found.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "published with per-message options set")]
async fn defaults_asserted_on_a_stepped_publish_name_the_options() {
    let tb = one_delivery().await;

    tb.out::<Ledger>().assert_called(2).assert_options_default();
}

// --8<-- [start:transform]
/// A transform that writes this broker's per-message settings. It names the options type, so it
/// mounts over this broker's publisher and nowhere else; the position it reads is still free, so
/// it goes on a reply and on a slot alike.
struct Expedite;

impl<K: ContextKind> PublishTransform<K, PriorityOptions> for Expedite {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PriorityOptions>,
        _cx: &K::View<'_>,
    ) {
        options.get_or_insert_with(PriorityOptions::default).ttl = Some(2);
    }
}
// --8<-- [end:transform]

/// A transform that writes a field the call site also writes, to show which one wins.
struct Downgrade;

impl PublishTransform<ForSlot, PriorityOptions> for Downgrade {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PriorityOptions>,
        _cx: &SlotContext<'_>,
    ) {
        options
            .get_or_insert_with(PriorityOptions::default)
            .priority = Some(1);
    }
}

/// The transform completes the call rather than replacing it: the call site names one field, the
/// transform another, and the broker resolves both against the policy's defaults.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_slot_transform_completes_what_the_call_site_set() {
    let app = RustStream::new(AppInfo::new("options-transform", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3).ttl(9))
                .transform(Expedite)
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");

    // Nothing at the call site: the transform's field applies, the other keeps the policy's.
    tb.broker::<MemoryBroker>()
        .published::<Order>("options.normal")
        .assert_called_once()
        .with_header("priority", "3")
        .with_header("ttl", "2");
    // A step at the call site: both fields are set, by different writers.
    tb.broker::<MemoryBroker>()
        .published::<Order>("options.urgent")
        .assert_called_once()
        .with_header("priority", "9")
        .with_header("ttl", "2");
    tb.out::<Ledger>()
        .assert_called(2)
        .with_options(&PriorityOptions {
            priority: Some(9),
            ttl: Some(2),
        });
}

/// A transform writes after the call site, so it wins a field they both name.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_slot_transform_overrides_a_field_the_call_site_named() {
    let app = RustStream::new(AppInfo::new("options-override", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(record)
                .out(Ledger, PriorityPublish::default().priority(3))
                .transform(Downgrade)
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 7 })
        .to("options.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Order>("options.urgent")
        .assert_called_once()
        .with_header("priority", "1");
}

/// A reply has no call site, so a transform is the only thing that adjusts its settings, and the
/// harness reads them back on the channel the reply went to.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_reply_transform_sets_what_the_reply_has_no_call_site_for() {
    let app = RustStream::new(AppInfo::new("options-reply-transform", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .transform(Expedite);
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 2 })
        .to("options.reply.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Receipt>("options.reply.out")
        .assert_called_once()
        .with(&Receipt { id: 2 })
        .with_header("priority", "4")
        .with_header("ttl", "2")
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

#[subscriber("options.batch.in", reply("options.batch.out"))]
async fn confirm_all(orders: &[Order]) -> Vec<Receipt> {
    orders
        .iter()
        .map(|order| Receipt { id: order.id })
        .collect()
}

/// The batch path carries the position too: the same transform, lifted with `for_batch`, writes
/// the settings of every reply in the batch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_batch_transform_sets_the_options_of_every_reply() {
    let app = RustStream::new(AppInfo::new("options-batch", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(confirm_all.batch(nonzero!(8)))
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .batch_transform(for_batch(Expedite));
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.message(&Order { id: 3 })
        .to("options.batch.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Receipt>("options.batch.out")
        .assert_called_once()
        .with(&Receipt { id: 3 })
        .with_header("priority", "4")
        .with_header("ttl", "2")
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

#[derive(OutSlot)]
#[publishes(Receipt)]
struct Followups;

/// Publishes a receipt to the channel `acknowledge` replies on, at a priority of its own.
#[subscriber("options.follow.in")]
async fn follow_up(
    order: &Order,
    Out(followups): Out<impl Publisher<Options = PriorityOptions>, Followups>,
) -> HandlerOutcome {
    if followups
        .message(&Receipt { id: order.id })
        .to("options.reply.out")
        .priority(9)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// A reply with options, then a slot publish to the same channel with other options: the
/// channel view reads the options of the latest publish, whatever made it.
async fn reply_then_slot_publish() -> TestApp<()> {
    let app = RustStream::new(AppInfo::new("options-latest", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .transform(Expedite);
            b.include(follow_up)
                .out(Followups, PriorityPublish::default().priority(3))
                .build();
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");
    tb.message(&Order { id: 2 })
        .to("options.reply.in")
        .publish()
        .await
        .expect("publish");
    tb.message(&Order { id: 5 })
        .to("options.follow.in")
        .publish()
        .await
        .expect("publish");
    tb
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_channel_view_reads_the_options_of_the_latest_publish() {
    let tb = reply_then_slot_publish().await;

    tb.broker::<MemoryBroker>()
        .published::<Receipt>("options.reply.out")
        .assert_called(2)
        .with(&Receipt { id: 5 })
        .with_options(&PriorityOptions {
            priority: Some(9),
            ttl: None,
        });
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "unexpected per-message options")]
async fn the_options_of_an_earlier_reply_do_not_stand_for_a_later_publish() {
    let tb = reply_then_slot_publish().await;

    tb.broker::<MemoryBroker>()
        .published::<Receipt>("options.reply.out")
        .assert_called(2)
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

/// The test's own publish reaches no broker's options type, so a reply's options before it do
/// not answer for it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "the test published")]
async fn the_options_of_a_reply_do_not_stand_for_the_tests_own_publish() {
    let app = RustStream::new(AppInfo::new("options-own", "0.1.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .transform(Expedite);
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");
    tb.message(&Order { id: 2 })
        .to("options.reply.in")
        .publish()
        .await
        .expect("publish");
    tb.message(&Receipt { id: 99 })
        .to("options.reply.out")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Receipt>("options.reply.out")
        .assert_called(2)
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

/// The same policy on a broker that retains its log.
impl PublishPolicy<ConnectedMemoryBroker<Retaining>> for PriorityPublish {
    type Live = PriorityPublisher;

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

/// A broker that keeps one message per channel: the test's own publish of the reply's bytes
/// evicts the reply, and the reply's options do not pass to the message that replaced it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "the test published")]
async fn the_options_of_an_evicted_reply_do_not_stand_for_the_tests_own_publish() {
    let app = RustStream::new(AppInfo::new("options-evicted", "0.1.0")).with_broker(
        MemoryBroker::retaining(Retention::Messages(nonzero!(1))),
        |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .transform(Expedite);
        },
    );
    let tb = TestApp::start(app).await.expect("harness start");
    tb.message(&Order { id: 2 })
        .to("options.reply.in")
        .publish()
        .await
        .expect("publish");
    tb.message(&Receipt { id: 2 })
        .to("options.reply.out")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker<Retaining>>()
        .published::<Receipt>("options.reply.out")
        .assert_called_once()
        .with(&Receipt { id: 2 })
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

/// Two brokers reply on a channel of the same name, one with a transform that sets options and
/// one without: each broker's view reads its own replies' options.
async fn two_brokers_one_channel_name() -> TestApp<()> {
    let app = RustStream::new(AppInfo::new("options-brokers", "0.1.0"))
        .with_broker_labeled("expedited", MemoryBroker::new(), |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9))
                .transform(Expedite);
        })
        .with_broker_labeled("plain", MemoryBroker::new(), |b| {
            b.include(acknowledge)
                .out_reply(PriorityPublish::default().priority(4).ttl(9));
        });
    let tb = TestApp::start(app).await.expect("harness start");
    for label in ["plain", "expedited"] {
        tb.broker_named(label)
            .message(&Order { id: 2 })
            .to("options.reply.in")
            .publish()
            .await
            .expect("publish");
    }
    tb
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn each_broker_reads_the_options_of_its_own_publishes() {
    let tb = two_brokers_one_channel_name().await;

    tb.broker_named("plain")
        .published::<Receipt>("options.reply.out")
        .assert_called_once()
        .assert_options_default();
    tb.broker_named("expedited")
        .published::<Receipt>("options.reply.out")
        .assert_called_once()
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[should_panic(expected = "published with the policy's default options")]
async fn another_brokers_options_do_not_stand_for_this_brokers_publish() {
    let tb = two_brokers_one_channel_name().await;

    tb.broker_named("plain")
        .published::<Receipt>("options.reply.out")
        .assert_called_once()
        .with_options(&PriorityOptions {
            priority: None,
            ttl: Some(2),
        });
}