ruststream-sea-file 0.7.1

//! Standard input and output as one stream: a service that is a stage of a shell pipeline.
//!
//! [`StdioBroker`] records nothing and attaches the process's own pipes on
//! [`Broker::connect`]. Standard input is the subscription, standard output the publisher, so
//! `producer | service | consumer` works with ordinary command-line tools. A service on a
//! pipeline globs this form's [`prelude`] and names its policy [`Publish`].
//!
//! - [StdioBroker][stdio-broker]: [`StdioBroker`], and [`loopback`](StdioBroker::loopback) for a
//!   stage that reads its own output ([loopback][loopback]).
//! - [Lines][lines]: the client's `[timestamp | stream_key | sequence | shard_id] payload` line
//!   format, read by a subscription on its stream key through [`StdioSubscriber`].
//! - [StdioPublish][stdio-publish]: [`Publish`] and its live [`StdioPublisher`], one line per
//!   message on standard output.
//! - [Payloads][payloads]: binary, multi-line and padded payloads through the envelope.
//! - [out_retry][out-retry]: where a deferred copy goes, named at the mount site.
//! - [Shutdown][shutdown]: a stdio shutdown ends every stdio stream of the process.
//!
//! [stdio-broker]: https://powersemmi.github.io/ruststream-sea-file/latest/components/broker/stdio-broker/
//! [loopback]: https://powersemmi.github.io/ruststream-sea-file/latest/components/broker/loopback/
//! [lines]: https://powersemmi.github.io/ruststream-sea-file/latest/components/subscriptions/lines/
//! [stdio-publish]: https://powersemmi.github.io/ruststream-sea-file/latest/components/publishing/stdio-publish/
//! [payloads]: https://powersemmi.github.io/ruststream-sea-file/latest/components/headers/payloads/
//! [out-retry]: https://powersemmi.github.io/ruststream-sea-file/latest/components/settlement/out-retry/
//! [shutdown]: https://powersemmi.github.io/ruststream-sea-file/latest/components/broker/shutdown/

// Without the `testing` feature a link has one variant, so a `match` on it has a single arm; the
// matches stay so that the in-process arm has its place when the feature is on.
#![cfg_attr(
    not(feature = "testing"),
    allow(clippy::infallible_destructuring_match)
)]

use std::future::{Future, ready};
use std::io;
use std::num::NonZeroUsize;
use std::pin::pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use futures::{FutureExt as _, Stream};
#[cfg(feature = "testing")]
use ruststream::RawMessage;
#[cfg(feature = "testing")]
use ruststream::testing::{Coordinator, InProcess, TestableBroker};
use ruststream::{
    BatchSubscriber, Broker, BufferedSubscriber, ConnectedBroker, DefaultPublish, DescribeServer,
    Lend, NamedCopies, OutgoingMessage, PairError, PublishPolicy, Publisher, ServerSpec, Subscribe,
    Subscriber,
};
use sea_streamer_stdio::{
    StdioConnectOptions, StdioConsumer, StdioConsumerOptions, StdioProducer, StdioProducerOptions,
    StdioResult, StdioStreamer,
};
use sea_streamer_types::{
    Consumer as _, ConsumerMode, ConsumerOptions as _, Producer as _, StreamKey, Streamer as _,
    StreamerUri,
};
use tokio::runtime::Handle;
use tokio::sync::{OnceCell, mpsc};

use crate::batching::BATCH_MAX_WAIT;
use crate::error::{SeaFileError, box_err};
#[cfg(feature = "testing")]
use crate::in_process::{MemoryPipe, PipeQueue};
use crate::message::SeaMessage;
use crate::wire;

/// Whether a subscription in this process has started the client's standard-input reader.
///
/// Why a static: the reader is the client's, and it is process-wide. The first consumer starts
/// it, and it then holds standard input while it waits for a line, so only the subscription that
/// starts it can hold standard input itself (see `open_consumer`); a later one would wait for
/// the next line.
static READER_STARTED: AtomicBool = AtomicBool::new(false);

pub(crate) struct StdioCore {
    pub(crate) link: StdioLink,
    pub(crate) closed: AtomicBool,
    /// The broker's own setting, which the test harness reads to know where a publish goes. The
    /// field is there only with the `testing` feature.
    #[cfg(feature = "testing")]
    loopback: bool,
}

/// What a connected broker and every handle paired off it speak over: the process's own pipes,
/// or, under the `testing` feature, the in-memory ones the test harness connected instead.
///
/// Without the feature there is one variant, so the type is the client's streamer itself and every
/// `match` on it is irrefutable: a production build carries no second transport and no branch to
/// it.
pub(crate) enum StdioLink {
    Stdio(Pipes),
    #[cfg(feature = "testing")]
    InProcess(Arc<MemoryPipe>),
}

/// The process's own pipes as the client opened them, and the runtime `connect` ran on: the
/// reader task of every subscription runs there, whichever runtime opens it.
pub(crate) struct Pipes {
    streamer: StdioStreamer,
    runtime: Handle,
}

// The zero-cost promise of the in-process mode, held by the compiler: a build without it gives the
// link exactly the size of the pipes it wraps.
#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<StdioLink>() == size_of::<Pipes>());

impl StdioCore {
    fn ensure_open(&self) -> Result<(), SeaFileError> {
        if self.closed.load(Ordering::Acquire) {
            return Err(SeaFileError::NotConnected);
        }
        Ok(())
    }
}

impl std::fmt::Debug for StdioCore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StdioCore")
            .field("closed", &self.closed.load(Ordering::Relaxed))
            .finish_non_exhaustive()
    }
}

type StdioCell = Arc<OnceCell<Arc<StdioCore>>>;

/// Standard input and output as one stream: consume lines from stdin, publish lines to
/// stdout, in the client's `[timestamp | stream_key | seq] payload` line format.
///
/// Shutting this broker down ends every stdio subscription and publisher in the process, after
/// the lines each subscription has already read: the client's transport is process-wide. The
/// [StdioBroker](https://powersemmi.github.io/ruststream-sea-file/latest/components/broker/stdio-broker/)
/// page shows it in a service.
///
/// # Examples
///
/// ```
/// use ruststream_sea_file::stdio::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Deserialize)]
/// struct Line {
///     text: String,
/// }
///
/// #[derive(Debug, Outgoing, Serialize)]
/// #[outgoing(name = "upper")]
/// struct Upper {
///     text: String,
/// }
///
/// #[subscriber("lines", reply)]
/// async fn shout(line: &Line) -> Upper {
///     Upper { text: line.text.to_uppercase() }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("shout", "0.1.0")).with_broker(StdioBroker::new(), |b| {
///         b.include(shout).out_retry(Publish).to("lines.retry");
///     })
/// }
/// ```
#[derive(Debug, Clone, Default)]
#[must_use]
pub struct StdioBroker {
    loopback: bool,
    cell: StdioCell,
}

impl StdioBroker {
    /// Records configuration only. No I/O.
    pub fn new() -> Self {
        Self::default()
    }

    /// Loops published messages back to this process's own subscribers (for tests).
    pub fn loopback(mut self) -> Self {
        self.loopback = true;
        self
    }

    /// A publisher sharing this broker's connection cell; buildable before `connect`.
    #[must_use]
    pub fn publisher(&self) -> StdioPublisher {
        StdioPublisher {
            cell: Arc::clone(&self.cell),
            producer: Arc::new(OnceCell::new()),
        }
    }
}

impl Broker for StdioBroker {
    type Error = SeaFileError;
    type Connected = ConnectedStdioBroker;

    async fn connect(self) -> Result<Self::Connected, Self::Error> {
        let core = self
            .cell
            .get_or_try_init(async || {
                let mut options = StdioConnectOptions::default();
                options.set_loopback(self.loopback);
                let streamer = StdioStreamer::connect(StreamerUri::zero(), options)
                    .await
                    .map_err(|e| SeaFileError::Connect {
                        target: "stdio".to_owned(),
                        source: box_err(e),
                    })?;
                Ok::<_, SeaFileError>(Arc::new(StdioCore {
                    link: StdioLink::Stdio(Pipes {
                        streamer,
                        runtime: Handle::current(),
                    }),
                    closed: AtomicBool::new(false),
                    #[cfg(feature = "testing")]
                    loopback: self.loopback,
                }))
            })
            .await?
            .clone();
        Ok(ConnectedStdioBroker {
            core,
            cell: self.cell,
        })
    }
}

/// The in-process mode: the connected form a test runs the production app against, over a pipe
/// held in memory in place of the process's own standard input and output.
///
/// The pipe belongs to this broker and every clone of it. What the test injects arrives on its
/// standard input; what the service publishes is written to its standard output, and reaches the
/// service's own subscriptions only under [`loopback`](Self::loopback), as on a real pipe.
#[cfg(feature = "testing")]
impl InProcess for StdioBroker {
    async fn connect_in_process(self) -> Result<Self::Connected, Self::Error> {
        let core = self
            .cell
            .get_or_init(async || {
                Arc::new(StdioCore {
                    link: StdioLink::InProcess(MemoryPipe::new(self.loopback)),
                    closed: AtomicBool::new(false),
                    loopback: self.loopback,
                })
            })
            .await
            .clone();
        Ok(ConnectedStdioBroker {
            core,
            cell: self.cell,
        })
    }
}

#[cfg(feature = "testing")]
ruststream::register_testable_broker!(StdioBroker);

/// Opens a consumer of `key` on standard input.
///
/// The client starts its reader inside the first consumer and only then registers that consumer,
/// so a line already waiting on the pipe could be read in between and handed to no one: a stage
/// started with its input ready lost its first line. The first subscription therefore holds
/// standard input while the consumer is created, and the reader takes its first line only once
/// the consumer is there. Creating a consumer completes without waiting, so standard input is
/// held for that one poll and never across an await.
async fn open_consumer(streamer: &StdioStreamer, key: StreamKey) -> StdioResult<StdioConsumer> {
    let keys = [key];
    let mut created =
        pin!(streamer.create_consumer(&keys, StdioConsumerOptions::new(ConsumerMode::RealTime)));
    if READER_STARTED.swap(true, Ordering::AcqRel) {
        return created.await;
    }
    let first = {
        let _input = io::stdin().lock();
        created.as_mut().now_or_never()
    };
    match first {
        Some(consumer) => consumer,
        None => created.await,
    }
}

impl DescribeServer for StdioBroker {
    fn describe_server(&self) -> ServerSpec {
        ServerSpec::in_process("stdio")
    }
}

/// The typed witness that `connect` succeeded.
#[derive(Debug)]
pub struct ConnectedStdioBroker {
    core: Arc<StdioCore>,
    // Keeps the cell of publishers handed out before connect alive and filled.
    cell: StdioCell,
}

impl ConnectedStdioBroker {
    /// A publisher from the connected form.
    #[must_use]
    pub fn publisher(&self) -> StdioPublisher {
        StdioPublisher {
            cell: Arc::clone(&self.cell),
            producer: Arc::new(OnceCell::new()),
        }
    }
}

impl ConnectedBroker for ConnectedStdioBroker {
    type Error = SeaFileError;
    type Closed = ();

    async fn shutdown(self) -> Result<(), Self::Error> {
        self.core.closed.store(true, Ordering::Release);
        let streamer = match &self.core.link {
            StdioLink::Stdio(pipes) => &pipes.streamer,
            // The in-memory pipe belongs to this broker alone, so its shutdown ends only the
            // subscriptions on it.
            #[cfg(feature = "testing")]
            StdioLink::InProcess(pipe) => {
                pipe.finish();
                return Ok(());
            }
        };
        // Globally destructive by the client's design: every stdio consumer and producer in
        // the process ends. That is the honest meaning of shutting down a process-wide
        // transport.
        streamer
            .clone()
            .disconnect()
            .await
            .map_err(|e| SeaFileError::Connect {
                target: "stdio".to_owned(),
                source: box_err(e),
            })
    }
}

impl Subscribe for ConnectedStdioBroker {
    type Subscriber = StdioSubscriber;
    /// The mount site names where a deferred copy goes, because nothing on this transport
    /// addresses the subscription: a publish writes to standard output and a subscription reads
    /// standard input, so the process downstream of the pipe is not the one that sent the
    /// message.
    ///
    /// A registration over stdio therefore names a destination - `.out_retry(Publish).to(name)`,
    /// or a transform that names one per delivery - and one that names neither refuses to start,
    /// which is the right answer for a pipeline stage: a delayed message that went nowhere is
    /// worse than a service that does not come up. ([`loopback`](StdioBroker::loopback) does
    /// route a publish back to this process, but it is a test aid, and an address that only held
    /// under it would break in the shape a service ships.)
    type Copies = NamedCopies;

    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error> {
        self.core.ensure_open()?;
        let key =
            StreamKey::new(name).map_err(|e| SeaFileError::Invalid(format!("'{name}': {e}")))?;
        let pipes = match &self.core.link {
            StdioLink::Stdio(pipes) => pipes,
            #[cfg(feature = "testing")]
            StdioLink::InProcess(pipe) => {
                return Ok(StdioSubscriber {
                    stream: name.to_owned(),
                    inner: BufferedSubscriber::new(StdioDeliveries::InProcess(
                        pipe.subscribe(key.name()),
                    ))
                    .max_wait(BATCH_MAX_WAIT),
                });
            }
        };
        let consumer =
            open_consumer(&pipes.streamer, key)
                .await
                .map_err(|e| SeaFileError::Subscribe {
                    stream: name.to_owned(),
                    source: box_err(e),
                })?;

        let (tx, rx) = mpsc::channel(64);
        pipes.runtime.spawn(async move {
            loop {
                tokio::select! {
                    () = tx.closed() => break,
                    next = consumer.next() => match next {
                        Ok(message) => {
                            if tx.send(Ok(SeaMessage::new(&message))).await.is_err() {
                                break;
                            }
                        }
                        // The client fails a read only once its consumers are disconnected,
                        // which is what the shutdown does: the subscription ends there.
                        Err(_) => break,
                    },
                }
            }
        });
        Ok(StdioSubscriber {
            stream: name.to_owned(),
            inner: BufferedSubscriber::new(StdioDeliveries::Reader(rx)).max_wait(BATCH_MAX_WAIT),
        })
    }
}

impl DefaultPublish for ConnectedStdioBroker {
    type Policy = StdioPublish;
}

/// The harness's view of the in-process transport: what it feeds to standard input, what it
/// reads back, and the coordinator it counts the in-flight deliveries with.
///
/// An injected message is a line arriving on standard input, so every subscription on its stream
/// key reads it. A publish is a line on standard output, which reaches the process downstream:
/// [`routes`](TestableBroker::routes) answers that it reaches this service's subscriptions of the
/// same stream key under [`loopback`](StdioBroker::loopback), and none of them otherwise.
///
/// # Panics
///
/// `inject` and `published` panic on a broker connected with `connect`: the harness drives only
/// the pipe `connect_in_process` produces. `inject` also panics on a line a pipe cannot carry: an
/// invalid stream key, or a message with no payload and no headers.
#[cfg(feature = "testing")]
impl TestableBroker for ConnectedStdioBroker {
    fn install_coordinator(&self, coordinator: Coordinator) {
        if let StdioLink::InProcess(pipe) = &self.core.link {
            pipe.install(coordinator);
        }
    }

    fn inject(&self, message: OutgoingMessage<'_>) {
        let pipe = self.memory_pipe("inject");
        let key = StreamKey::new(message.name()).unwrap_or_else(|err| {
            panic!(
                "the injected message to {:?} is not one a pipe carries: {err}",
                message.name()
            )
        });
        assert!(
            !(message.payload().is_empty() && message.headers().is_empty()),
            "the injected message to {:?} is empty, and a pipe drops an empty line",
            message.name(),
        );
        // The line an upstream stage of the pipeline writes for this message.
        pipe.feed(
            key,
            wire::encode(message.headers(), message.payload(), true),
        );
    }

    fn published(&self, name: &str) -> Vec<RawMessage> {
        self.memory_pipe("published").published(name)
    }

    fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize> {
        if !self.core.loopback {
            return Vec::new();
        }
        subscriptions
            .iter()
            .enumerate()
            .filter(|(_, name)| **name == destination)
            .map(|(position, _)| position)
            .collect()
    }
}

#[cfg(feature = "testing")]
impl ConnectedStdioBroker {
    /// The in-memory pipe, which is all the harness drives.
    fn memory_pipe(&self, what: &str) -> &MemoryPipe {
        match &self.core.link {
            StdioLink::InProcess(pipe) => pipe,
            StdioLink::Stdio(_) => panic!(
                "TestableBroker::{what} reached a broker connected with `connect`; the harness \
                 drives the pipe `connect_in_process` produces"
            ),
        }
    }
}

/// A subscription to one stream key on standard input; yields [`SeaMessage`]s.
///
/// Standard input has no retained log: there is no acknowledgement and no repositioning, and
/// both are reported as unsupported rather than pretended. Batches it does serve: the client
/// reads one line at a time, so they are assembled here, capped at the size the mount site
/// named.
pub struct StdioSubscriber {
    stream: String,
    inner: BufferedSubscriber<StdioDeliveries>,
}

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

impl Subscriber for StdioSubscriber {
    type Message = SeaMessage;
    type Error = SeaFileError;

    fn stream(&mut self) -> impl Stream<Item = Result<SeaMessage, SeaFileError>> + Send + '_ {
        self.inner.stream()
    }
}

impl BatchSubscriber for StdioSubscriber {
    type Batch = Vec<SeaMessage>;

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

/// A subscription's lines, before batching: one line per poll, in arrival order.
///
/// Without the `testing` feature there is one variant, so the type is the reader task's channel
/// itself and every `match` on it is irrefutable.
enum StdioDeliveries {
    /// The reader task's channel, fed from standard input.
    Reader(mpsc::Receiver<Result<SeaMessage, SeaFileError>>),
    /// The in-process pipe's queue.
    #[cfg(feature = "testing")]
    InProcess(PipeQueue),
}

// The zero-cost promise of the in-process mode, held by the compiler.
#[cfg(not(feature = "testing"))]
const _: () = assert!(
    size_of::<StdioDeliveries>() == size_of::<mpsc::Receiver<Result<SeaMessage, SeaFileError>>>()
);

impl Subscriber for StdioDeliveries {
    type Message = SeaMessage;
    type Error = SeaFileError;

    fn stream(&mut self) -> impl Stream<Item = Result<SeaMessage, SeaFileError>> + Send + '_ {
        // Poll the channel in place rather than wrapping it in an owning stream, so `stream`
        // can be called again after the returned stream is dropped (the runtime and the
        // conformance helpers re-enter it per call).
        futures::stream::poll_fn(move |cx| match self {
            Self::Reader(rx) => rx.poll_recv(cx),
            #[cfg(feature = "testing")]
            Self::InProcess(queue) => queue.poll_next(cx).map(|next| next.map(Ok)),
        })
    }
}

/// Publishes messages to standard output.
///
/// The line format is the client's own. A line is text, ends at a newline, and the reader trims
/// it, so a payload that is not UTF-8, holds a newline or begins or ends with whitespace (and any
/// message with headers) travels in the text-safe envelope. The client silently drops
/// empty lines, so an empty payload is rejected here instead.
///
/// A publish carries no per-message settings: a line on standard output takes a key and a
/// payload and nothing else, so [`Publisher::Options`] is the unit type and the publish builder
/// gains no step from this crate.
///
/// Standard output reaches the next process in the pipeline and never this process's own
/// standard input, so a registration that defers a delivery names where the copy goes, with
/// `.out_retry(Publish).to(key)` or a transform that names one per delivery; one that names
/// neither refuses to start. The
/// [StdioPublish](https://powersemmi.github.io/ruststream-sea-file/latest/components/publishing/stdio-publish/)
/// page shows it in a service.
///
/// # Examples
///
/// ```
/// use ruststream_sea_file::stdio::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Deserialize)]
/// struct Job {
///     id: u64,
/// }
///
/// #[derive(Debug, Outgoing, Serialize)]
/// #[outgoing(name = "results")]
/// struct Done {
///     id: u64,
/// }
///
/// #[subscriber("jobs", reply)]
/// async fn work(job: &Job) -> Done {
///     tracing::info!(job.id, "done");
///     Done { id: job.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("pipeline", "0.1.0")).with_broker(StdioBroker::new(), |b| {
///         b.include(work).out_retry(Publish).to("jobs.retry");
///     })
/// }
/// ```
#[derive(Clone)]
pub struct StdioPublisher {
    cell: StdioCell,
    producer: Arc<OnceCell<StdioProducer>>,
}

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

impl Publisher for StdioPublisher {
    /// A line on standard output is written from the bytes and keeps nothing: the client's
    /// `send_to` takes a slice, and the text-safe envelope is a buffer of this crate's own.
    type Payload = Lend;

    type Error = SeaFileError;
    type Options = ();

    async fn publish(
        &self,
        msg: OutgoingMessage<'_, &[u8]>,
        _options: Option<&()>,
    ) -> Result<(), Self::Error> {
        let core = self.cell.get().ok_or(SeaFileError::NotConnected)?;
        core.ensure_open()?;
        if msg.payload().is_empty() && msg.headers().is_empty() {
            return Err(SeaFileError::Invalid(
                "stdio drops empty lines; an empty message cannot be transmitted".into(),
            ));
        }
        let streamer = match &core.link {
            StdioLink::Stdio(pipes) => &pipes.streamer,
            #[cfg(feature = "testing")]
            StdioLink::InProcess(pipe) => {
                let key = StreamKey::new(msg.name())
                    .map_err(|e| SeaFileError::Invalid(format!("'{}': {e}", msg.name())))?;
                pipe.write(key, wire::encode(msg.headers(), msg.payload(), true));
                return Ok(());
            }
        };
        let producer = self
            .producer
            .get_or_try_init(async || {
                streamer
                    .create_generic_producer(StdioProducerOptions::default())
                    .await
                    .map_err(|e| SeaFileError::Publish {
                        stream: msg.name().to_owned(),
                        source: box_err(e),
                    })
            })
            .await?;
        let key = StreamKey::new(msg.name())
            .map_err(|e| SeaFileError::Invalid(format!("'{}': {e}", msg.name())))?;
        // force_text: the stdio line format rejects non-UTF-8 payloads.
        let payload = wire::encode(msg.headers(), msg.payload(), true);
        producer
            .send_to(&key, payload.as_slice())
            .map_err(|e| SeaFileError::Publish {
                stream: msg.name().to_owned(),
                source: box_err(e),
            })?
            .await
            .map(|_| ())
            .map_err(|e| SeaFileError::Publish {
                stream: msg.name().to_owned(),
                source: box_err(e),
            })
    }
}

/// The publish policy for [`StdioPublisher`].
///
/// # Examples
///
/// ```
/// use ruststream_sea_file::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Debug, Outgoing, Serialize)]
/// struct Receipt {
///     order: u64,
/// }
///
/// #[subscriber("orders", reply("receipts"))]
/// async fn confirm(order: &Order) -> Receipt {
///     Receipt { order: order.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(StdioBroker::new(), |b| {
///         b.include(confirm)
///             .out_reply(StdioPublish)
///             .out_retry(StdioPublish)
///             .to("orders.retry");
///     })
/// }
/// ```
#[derive(Debug, Clone, Copy, Default)]
#[must_use]
pub struct StdioPublish;

impl PublishPolicy<ConnectedStdioBroker> for StdioPublish {
    type Live = StdioPublisher;

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

/// The publish policy of this form, under the name every form uses.
///
/// A mount site names the concept, never the transport: moving a service from one form to
/// another changes the prelude it globs and leaves the composition root alone. The prefixed
/// [`StdioPublish`] stays at the crate root, for a service that mixes both forms.
///
/// # Examples
///
/// ```
/// use ruststream_sea_file::stdio::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Debug, Outgoing, Serialize)]
/// struct Receipt {
///     order: u64,
/// }
///
/// #[subscriber("orders", reply("receipts"))]
/// async fn confirm(order: &Order) -> Receipt {
///     Receipt { order: order.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(StdioBroker::new(), |b| {
///         b.include(confirm)
///             .out_reply(Publish)
///             .out_retry(Publish)
///             .to("orders.retry");
///     })
/// }
/// ```
pub use StdioPublish as Publish;

pub mod prelude {
    //! The imports a service on a shell pipeline writes every time, in one glob.
    //!
    //! The framework's prelude, this form's broker, and [`Publish`]. A subscription here is a
    //! plain stream key, so this form has no descriptor type.
    //!
    //! This is the routes-side vocabulary: a mount site globs it and names policies by concept.
    //! A handler body globs `ruststream::prelude` instead and bounds an injected publisher by
    //! the framework's capability traits, so the two vocabularies never meet in one file.
    //!
    //! # Examples
    //!
    //! ```
    //! use ruststream_sea_file::stdio::prelude::*;
    //! use serde::{Deserialize, Serialize};
    //!
    //! #[derive(Debug, Deserialize)]
    //! struct Job {
    //!     id: u64,
    //! }
    //!
    //! #[derive(Debug, Outgoing, Serialize)]
    //! #[outgoing(name = "results")]
    //! struct Done {
    //!     id: u64,
    //! }
    //!
    //! #[subscriber("jobs", reply)]
    //! async fn work(job: &Job) -> Done {
    //!     Done { id: job.id }
    //! }
    //!
    //! #[ruststream::app]
    //! fn app() -> impl App {
    //!     RustStream::new(AppInfo::new("pipeline", "0.1.0"))
    //!         .with_broker(StdioBroker::new(), |b| {
    //!             b.include(work).out_retry(Publish).to("jobs.retry");
    //!         })
    //! }
    //! ```

    pub use ruststream::prelude::*;

    // No capability traits: `SeaMessage` implements `Positioned`, but a pipe cannot seek back to
    // one - do not add it here.
    pub use crate::stdio::{Publish, StdioBroker};
}