ruststream 0.7.5

//! The type-selected pipeline lanes end to end: `Deserialize`/`Serialize` mean the framework's
//! codec does the work, `Deserialized`/`Serialized` mean the user's own type already did it -
//! and one signature mixes the lanes freely, because each end picks its own.
#![cfg(all(
    feature = "memory",
    feature = "macros",
    feature = "json",
    feature = "testing"
))]

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

/// A self-deserializing view over the payload: the framework's codec never runs on this lane.
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);

#[derive(
    Debug, Outgoing, serde::Serialize, serde::Deserialize, schemars::JsonSchema, PartialEq,
)]
struct Report {
    len: usize,
}

/// Bytes injected as themselves: what the tests feed the `Deserialized` lanes, where the payload
/// is deliberately not a model. It declares no name, so each injection names its subject - unlike
/// [`WireExport`], whose declaration fixes one.
#[derive(Outgoing, Serialized)]
struct Wire(Vec<u8>);

/// Raw in, encoded out: a `Deserialized` input composes with a `Serialize` reply - the input
/// lane skips the codec, the reply still rides it.
#[subscriber("lanes.measure", reply("lanes.measure.out"))]
async fn measure(frame: &Frame<'_>) -> Report {
    Report { len: frame.0.len() }
}

/// The lanes are picked per end: a raw input skips the codec while its reply still rides it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_raw_input_answers_with_an_encoded_reply() {
    let app =
        RustStream::new(AppInfo::new("lanes", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(measure);
        });
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&Wire(b"four".to_vec()))
        .to("lanes.measure")
        .publish()
        .await
        .expect("publish");

    tb.broker::<MemoryBroker>()
        .published::<Report>("lanes.measure.out")
        .assert_called_once()
        .with(&Report { len: 4 });
}

// --8<-- [start:serialized_out]
/// A self-carrying model in a slot dictionary: `#[derive(Serialized)]` routes it onto the
/// serialized wire, `#[derive(Outgoing)]` declares where it goes, and `#[publishes(..)]` makes
/// it a member like any encoded model.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "lanes.exports")]
struct WireExport(Vec<u8>);

#[derive(OutSlot)]
#[publishes(WireExport)]
struct Exports;

/// One typed entry serves both wires: the type picks the lane, so `message(&wire)` publishes
/// the bytes as they are - no codec anywhere - to the destination the declaration names.
#[subscriber("lanes.chunks")]
async fn export(frame: &Frame<'_>, Out(out): Out<impl Publisher, Exports>) -> HandlerOutcome {
    let wire = WireExport(frame.0.to_vec());
    if out.message(&wire).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}
// --8<-- [end:serialized_out]

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_serialized_out_type_is_a_dictionary_member() {
    let broker = MemoryBroker::new();
    let ingress = broker.publisher();

    let app = RustStream::new(AppInfo::new("exports", "0.1.0")).with_broker(broker, |b| {
        b.include(export).out(Exports, Publish).build();
    });
    let tb = TestApp::start(app).await.expect("harness start");

    ingress
        .message(&Wire(b"chunk".to_vec()))
        .to("lanes.chunks")
        .publish()
        .await
        .expect("publish");
    tb.settle().await.expect("settle");

    tb.broker::<MemoryBroker>()
        .published::<Vec<u8>>("lanes.exports")
        .assert_called_once()
        .with_raw(b"chunk");

    tb.shutdown().await.expect("graceful shutdown");
}

/// The other end of the same wire: whatever the test client injects on it arrives as it was
/// sent, so the assertion below reads the delivery raw.
#[subscriber("lanes.exports")]
async fn absorb(frame: &Frame<'_>) -> HandlerOutcome {
    let _ = frame.0.len();
    HandlerOutcome::ack()
}

/// The test client is the service's publish path, so it owes the lanes the same parity: a
/// `Serialized` value injected with `message(..)` takes its destination from the declaration
/// (no `to(..)` in the call) and leaves the bytes alone (no codec between value and wire).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_test_client_injects_a_serialized_value_verbatim() {
    let app =
        RustStream::new(AppInfo::new("exports", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(absorb);
        });
    let tb = TestApp::start(app).await.expect("harness start");

    tb.broker::<MemoryBroker>()
        .message(&WireExport(vec![0xde, 0xad, 0xbe, 0xef]))
        .publish()
        .await
        .expect("inject");

    // An encoding step between the value and the wire would have rewritten these bytes.
    tb.broker::<MemoryBroker>()
        .subscriber("lanes.exports")
        .assert_called_once()
        .with_raw(b"\xde\xad\xbe\xef")
        .settled(HandlerOutcome::ack());

    tb.shutdown().await.expect("graceful shutdown");
}

/// The typed form reports the metadata the raw seam reported: the serialized member is
/// schema-free by design, listed under its own name, and never a coverage gap.
#[cfg(feature = "asyncapi")]
#[test]
fn the_typed_form_keeps_the_serialized_metadata() {
    use ruststream::asyncapi::build_spec;

    let app =
        RustStream::new(AppInfo::new("exports", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(export).out(Exports, Publish).build();
        });
    let spec = build_spec(&app);

    let message = &spec.components.messages["WireExport"];
    assert!(
        message.payload.is_none(),
        "a serialized wire has no serde model to report"
    );
    assert!(spec.channels.contains_key("lanes.exports"));
    assert!(
        spec.messages_without_schema().is_empty(),
        "the serialized wire is schema-free by design, not a gap"
    );
}