ruststream 0.7.2

//! Router parity: every definition form the broker scope mounts is reachable from a standalone
//! [`Router`], through the same form-token dispatch.
//!
//! The scope's own coverage of these forms lives next to each feature (`out_injection`,
//! `out_slots`, `seek_api`, ...); what is asserted here is that the router surface reaches them
//! too, with the terminals a consuming builder needs.
#![cfg(all(
    feature = "memory",
    feature = "macros",
    feature = "json",
    feature = "testing"
))]

mod common;

use common::{Event, Wire};

use ruststream::memory::prelude::*;
use ruststream::testing::TestApp;

/// The payload view the byte-level bodies below take: the delivery's bytes, borrowed.
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);

/// The reply those bodies return: its bytes leave on the wire as they are, at the subject the
/// mount site names.
#[derive(Outgoing, Serialized)]
struct Export(Vec<u8>);

// Out slots: the single-slot shorthand, named slots, and the batch counterpart.

#[subscriber("rp.out.in")]
async fn forward(event: &Event, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    if out
        .message(event)
        .to("rp.out.forwarded")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The one-slot shorthand on a router: `.out(DefaultSlot, policy)` binds the unnamed slot, and
/// `.build()` commits.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_mounts_a_single_out_slot() {
    let router = Router::<MemoryBroker>::new()
        .include(forward)
        .out(DefaultSlot, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-out", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 3 })
        .to("rp.out.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.out.forwarded")
        .assert_called_once()
        .with(&Event { id: 3 });
}

#[derive(OutSlot)]
#[publishes(Wire)]
struct Encoded;

#[derive(OutSlot)]
#[publishes(Wire)]
struct Audit;

#[subscriber("rp.slots.in")]
async fn transcode(
    chunk: &Frame<'_>,
    Out(encoded): Out<impl Publisher, Encoded>,
    Out(audit): Out<impl Publisher, Audit>,
) -> HandlerOutcome {
    if encoded
        .message(&Wire::of(chunk.0))
        .to("rp.slots.encoded")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    let receipt = chunk.0.len().to_be_bytes();
    if audit
        .message(&Wire::of(receipt))
        .to("rp.slots.audit")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// Named slots bind by marker in any order, and `.build()` is the terminal.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_binds_named_out_slots_by_marker() {
    // Deliberately bound in the opposite of the signature order.
    let router = Router::<MemoryBroker>::new()
        .include(transcode)
        .out(Audit, Publish)
        .out(Encoded, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-slots", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Wire::of(b"frame"))
        .to("rp.slots.in")
        .publish()
        .await
        .expect("publish");

    tb.out::<Encoded>().assert_called_once().with_raw(b"frame");
    tb.out::<Audit>().assert_called_once();
}

/// A serialized dictionary member the slot's typed entry publishes byte-for-byte, the scope
/// counterpart being `tests/lanes.rs`.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "rp.wire.out")]
struct WireCopy(Vec<u8>);

#[derive(OutSlot)]
#[publishes(WireCopy)]
struct Wires;

#[subscriber("rp.wire.in")]
async fn copy_out(frame: &Frame<'_>, Out(out): Out<impl Publisher, Wires>) -> HandlerOutcome {
    if out
        .message(&WireCopy(frame.0.to_vec()))
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The serialized wire of a slot's typed entry mounts from a router exactly as from the scope:
/// the bytes leave as they are, at the destination the type declares.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_publishes_a_serialized_message_through_a_slot() {
    let router = Router::<MemoryBroker>::new()
        .include(copy_out)
        .out(Wires, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-wire", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Wire::of(b"frame"))
        .to("rp.wire.in")
        .publish()
        .await
        .expect("publish");
    tb.settle().await.expect("settle");

    tb.out::<Wires>().assert_called_once().with_raw(b"frame");
    tb.broker::<MemoryBroker>()
        .published::<WireCopy>("rp.wire.out")
        .assert_called_once()
        .with_raw(b"frame");
}

#[subscriber("rp.batch.in")]
async fn forward_batch(events: &[Event], Out(out): Out<impl Publisher>) -> HandlerOutcome {
    for event in events {
        if out
            .message(event)
            .to("rp.batch.forwarded")
            .publish()
            .await
            .is_err()
        {
            return HandlerOutcome::retry();
        }
    }
    HandlerOutcome::ack()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_mounts_a_batch_out_slot() {
    let router = Router::<MemoryBroker>::new()
        .include(forward_batch.batch(nonzero!(64)))
        .out(DefaultSlot, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-batch", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 9 })
        .to("rp.batch.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.batch.forwarded")
        .assert_called_once()
        .with(&Event { id: 9 });
}

// A broker context key: the seek handle rides the delivery context, read by the Ctx extractor.

#[subscriber(MemorySource::new("rp.seek.in"))]
async fn rewind(event: &Event, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
    if event.id == 0 && seeker.seek(MemoryPosition::start()).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_mounts_a_seek_key_reader() {
    // A seek key needs a log to reposition over, so this router mounts on a retaining broker.
    let router = Router::<MemoryBroker<Retaining>>::new().include(rewind);
    let app = RustStream::new(AppInfo::new("rp-seek", "0.1.0")).with_broker(
        MemoryBroker::retaining(Retention::Messages(nonzero!(32))),
        |b| b.include_router(router),
    );
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker<Retaining>>()
        .message(&Event { id: 1 })
        .to("rp.seek.in")
        .publish()
        .await
        .expect("publish");
    tb.settle().await.expect("settle");
    tb.broker::<MemoryBroker<Retaining>>()
        .subscriber("rp.seek.in")
        .assert_called_once();
}

// The reply terminals: `.build()` takes the broker's default publish policy, on the encoded and
// the byte-for-byte form alike.

#[subscriber("rp.reply.in", reply("rp.reply.out"))]
async fn relay(event: &Event) -> Event {
    Event { id: event.id + 1 }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_defaults_the_reply_publisher_on_mount() {
    let router = Router::<MemoryBroker>::new().include(relay).build();
    let app = RustStream::new(AppInfo::new("rp-reply", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 1 })
        .to("rp.reply.in")
        .publish()
        .await
        .expect("publish");
    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.reply.out")
        .assert_called_once()
        .with(&Event { id: 2 });
}

#[subscriber("rp.raw.in", reply("rp.raw.out"))]
async fn echo_frame(frame: &Frame<'_>) -> Export {
    Export(frame.0.to_vec())
}

/// The byte-reply form: its reply leaves unencoded, so it is its own route on a router.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_mounts_the_byte_reply_form() {
    let router = Router::<MemoryBroker>::new().include(echo_frame).build();
    let app = RustStream::new(AppInfo::new("rp-raw", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Wire::of(b"frame"))
        .to("rp.raw.in")
        .publish()
        .await
        .expect("publish");
    tb.settle().await.expect("settle");

    tb.broker::<MemoryBroker>()
        .published::<Export>("rp.raw.out")
        .assert_called_once()
        .with_raw(b"frame");
}

// The two-attachment forms: a reply next to Out slots, single and batch.

#[subscriber("rp.gate.in", reply("rp.gate.reply"))]
async fn gate(event: &Event, Out(out): Out<impl Publisher>) -> Result<Event, HandlerOutcome> {
    if out
        .message(event)
        .to("rp.gate.audit")
        .publish()
        .await
        .is_err()
    {
        return Err(HandlerOutcome::retry());
    }
    Ok(Event { id: event.id + 1 })
}

/// The reply side defaults while the slot side is bound explicitly.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_composes_a_default_reply_with_out_slots() {
    let router = Router::<MemoryBroker>::new()
        .include(gate)
        .out(DefaultSlot, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-gate", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 7 })
        .to("rp.gate.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.gate.audit")
        .assert_called_once()
        .with(&Event { id: 7 });
    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.gate.reply")
        .assert_called_once()
        .with(&Event { id: 8 });
}

#[subscriber("rp.audit.in", reply("rp.audit.out"))]
async fn audited_relay(frame: &Frame<'_>, Out(audit): Out<impl Publisher>) -> Export {
    audit
        .message(&Wire::of(frame.0))
        .to("rp.audit.copy")
        .publish()
        .await
        .expect("the slot publisher is live");
    Export(frame.0.to_vec())
}

/// The byte-reply form with an `Out` slot, reply side defaulted.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_composes_a_byte_reply_with_out_slots() {
    let router = Router::<MemoryBroker>::new()
        .include(audited_relay)
        .out(DefaultSlot, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-audit", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Wire::of(b"frame"))
        .to("rp.audit.in")
        .publish()
        .await
        .expect("publish");
    tb.settle().await.expect("settle");

    tb.broker::<MemoryBroker>()
        .published::<Export>("rp.audit.out")
        .assert_called_once()
        .with_raw(b"frame");
    tb.broker::<MemoryBroker>()
        .published::<Export>("rp.audit.copy")
        .assert_called_once()
        .with_raw(b"frame");
}

#[subscriber("rp.ledger.in", reply("rp.ledger.receipts"))]
async fn settle_batch(
    events: &[Event],
    Out(out): Out<impl Publisher>,
) -> Result<Vec<Event>, HandlerOutcome> {
    let batch = Event {
        id: u64::try_from(events.len()).expect("a batch fits in u64"),
    };
    if out
        .message(&batch)
        .to("rp.ledger.batches")
        .publish()
        .await
        .is_err()
    {
        return Err(HandlerOutcome::retry());
    }
    Ok(events
        .iter()
        .map(|event| Event { id: event.id + 100 })
        .collect())
}

/// The batch two-attachment form, with the reply side named explicitly this time.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_composes_a_batch_reply_with_out_slots() {
    let router = Router::<MemoryBroker>::new()
        .include(settle_batch.batch(nonzero!(64)))
        .out(Reply, Publish)
        .out(DefaultSlot, Publish)
        .build();
    let app = RustStream::new(AppInfo::new("rp-ledger", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 7 })
        .to("rp.ledger.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.ledger.receipts")
        .assert_called_once()
        .with(&Event { id: 107 });
    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.ledger.batches")
        .assert_called_once()
        .with(&Event { id: 1 });
}

// Cross-broker tokens reach a router include site, exactly as they reach a scope's.

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_accepts_a_cross_broker_bind_token() {
    let egress_broker = MemoryBroker::new().bindable();
    let egress = egress_broker.bind(Publish);

    // The token names its own broker, so the registration order of the two scopes is irrelevant:
    // the slot pairs against the egress broker while the subscription lives on the ingress one.
    let router = Router::<MemoryBroker>::new()
        .include(forward)
        .out(DefaultSlot, egress)
        .build();
    let app = RustStream::new(AppInfo::new("rp-bridge", "0.1.0"))
        .with_broker_labeled("ingress", MemoryBroker::new(), |b| b.include_router(router))
        .with_broker_labeled("egress", egress_broker, |_b| {});
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker_named("ingress")
        .message(&Event { id: 5 })
        .to("rp.out.in")
        .publish()
        .await
        .expect("publish");
    tb.broker_named("egress")
        .published::<Event>("rp.out.forwarded")
        .assert_called_once()
        .with(&Event { id: 5 });
}

// The batch reply terminal, and the metadata every new route kind contributes.

#[subscriber("rp.batch.in", reply("rp.batch.out"))]
async fn bulk_relay(events: &[Event]) -> Vec<Event> {
    events
        .iter()
        .map(|event| Event { id: event.id + 1 })
        .collect()
}

/// A batch publishing form mounted with only its batch size takes the broker's own default
/// publish policy: naming the size seals the definition, so the reply wiring defaults.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_router_defaults_the_batch_reply_publisher_on_mount() {
    let router = Router::<MemoryBroker>::new()
        .include(bulk_relay.batch(nonzero!(64)))
        .build();
    let app = RustStream::new(AppInfo::new("rp-batch", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(router));
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Event { id: 1 })
        .to("rp.batch.in")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Event>("rp.batch.out")
        .assert_called_once()
        .with(&Event { id: 2 });
}

/// Every route kind contributes its registration metadata, in registration order: that list is
/// what the `AsyncAPI` document is generated from.
#[test]
fn every_new_route_kind_reports_its_metadata_in_registration_order() {
    let router = Router::<MemoryBroker>::new()
        .include(rewind)
        .include(echo_frame)
        .build()
        .include(relay)
        .build()
        .include(bulk_relay.batch(nonzero!(64)))
        .build()
        .include(forward)
        .out(DefaultSlot, Publish)
        .build()
        .include(forward_batch.batch(nonzero!(64)))
        .out(DefaultSlot, Publish)
        .build();

    let names: Vec<_> = router.handlers().into_iter().map(|m| m.name).collect();
    assert_eq!(
        names,
        [
            "rp.seek.in",
            "rp.raw.in",
            "rp.reply.in",
            "rp.batch.in",
            "rp.out.in",
            "rp.batch.in",
        ]
    );
}