pipecircuit-cli 0.0.6

use core::{
    cell::Cell,
    convert::Infallible,
    num::{NonZeroU32, NonZeroUsize},
};
use std::io::{self, Write};

use __CRATE___core as app_core;
use __CRATE___switchboards as app_routes;

use app_core::{
    domain_primitives::Activation,
    pipes::{Activator, Client},
    state_machines::latch::{Activate, Latch},
};
use app_routes::{ActivationProgram, activation};
use pipecircuit::{
    ConductionFailure, DriveFailure, Inline, InlineSubmission, InstalledPipe, SubmitWith,
    storage::warp_table::Capacity,
};
use pipecircuit_mio::{
    MioRegistrationId, MioWakerOpenFailure,
    host::{
        MioHost, MioHostConfiguration, MioHostCreationFailure, MioHostFailure,
        MioInstallationFailure, MioPipeSet, MioProgress,
    },
    spsc::{QueueReception, QueueSubmission, SpscCapacity, SpscFailure, SpscPipe, pair},
};

#[derive(MioPipeSet)]
#[mio_roster(activation)]
pub struct Native {
    client: InstalledPipe<Client, SpscPipe<Client, Activation, Activate>>,
    activator: InstalledPipe<Activator, Inline<Latch, Activator>>,
}

type SourceFailure = DriveFailure<
    SpscFailure,
    activation::SubmissionFailure<QueueSubmission<Activation>, Infallible>,
>;
type ConductFailure = ConductionFailure<
    io::Error,
    MioHostFailure<SourceFailure, (), <Native as MioPipeSet>::ClockFailure>,
>;

pub enum Failure {
    Host(MioHostCreationFailure),
    Wake(MioWakerOpenFailure),
    Installation(MioInstallationFailure),
    Submit(QueueSubmission<Activate>),
    Conduct(ConductFailure),
    Reply(QueueReception<Activation>),
    CapacityWake {
        reply: Activation,
        failure: io::Error,
    },
    Output {
        reply: Activation,
        failure: io::Error,
    },
}

impl core::fmt::Debug for Failure {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::Host(failure) => formatter.debug_tuple("Host").field(failure).finish(),
            Self::Wake(failure) => formatter.debug_tuple("Wake").field(failure).finish(),
            Self::Installation(failure) => formatter
                .debug_tuple("Installation")
                .field(failure)
                .finish(),
            Self::Submit(failure) => formatter.debug_tuple("Submit").field(failure).finish(),
            Self::Conduct(failure) => formatter.debug_tuple("Conduct").field(failure).finish(),
            Self::Reply(failure) => formatter.debug_tuple("Reply").field(failure).finish(),
            Self::CapacityWake { reply, failure } => formatter
                .debug_struct("CapacityWake")
                .field("reply", reply)
                .field("failure", failure)
                .finish(),
            Self::Output { reply, failure } => formatter
                .debug_struct("Output")
                .field("reply", reply)
                .field("failure", failure)
                .finish(),
        }
    }
}

fn run_with_writer(writer: &mut impl Write) -> Result<(), Failure> {
    let host = MioHost::try_open(MioHostConfiguration {
        source_capacity: Capacity::new(NonZeroU32::MIN),
        poll_capacity: NonZeroUsize::MIN,
        dispatch_quantum: NonZeroUsize::MIN,
    })
    .map_err(Failure::Host)?;
    let wake = host
        .try_open_waker(MioRegistrationId::new(1))
        .map_err(Failure::Wake)?;
    let (endpoint, mut client) = pair(
        SpscCapacity {
            to_peer: NonZeroUsize::MIN,
            from_peer: NonZeroUsize::MIN,
        },
        wake.clone(),
        wake,
    );
    let pipes = Native::from(activation::Pipes {
        client: InstalledPipe::new(SpscPipe::attach(endpoint)),
        activator: InstalledPipe::new(Inline::for_pipe(Latch::default())),
    });
    let replied = Cell::new(false);
    let policies =
        activation::SubmissionPolicies {
            client: SubmitWith(|submission| match submission {
                QueueSubmission::Admitted => {
                    replied.set(true);
                    Ok(())
                }
                refused @ (QueueSubmission::Rejected(_)
                | QueueSubmission::AdmittedButWakeFailed(_)) => Err(refused),
            }),
            activator: InlineSubmission,
        };
    let installation = pipes.with_submission_policies(policies, |source| match source {
        // The fixture submits exactly one request, producing one fixed-size reply.
        // Both queues reserve that slot. Streaming intake needs its own admission policy.
        NativeSourceTurn::Client(turn) => turn.conduct_unrestricted(|| {
            if replied.get() {
                MioProgress::Completed { completion: () }
            } else {
                MioProgress::Continue
            }
        }),
        NativeSourceTurn::Activator(turn) => match turn.activity {},
    });
    let mut pipecircuit = host
        .install_program(ActivationProgram, installation)
        .map_err(|refused| Failure::Installation(refused.reason))?;
    match client.submit(Activate) {
        QueueSubmission::Admitted => {}
        refused @ (QueueSubmission::Rejected(_) | QueueSubmission::AdmittedButWakeFailed(_)) => {
            return Err(Failure::Submit(refused));
        }
    }
    pipecircuit.conduct().map_err(Failure::Conduct)?;
    match client.receive() {
        QueueReception::Received {
            output,
            capacity_wake,
        } => {
            capacity_wake.map_err(|failure| Failure::CapacityWake {
                reply: output,
                failure,
            })?;
            writeln!(writer, "{output:?}").map_err(|failure| Failure::Output {
                reply: output,
                failure,
            })
        }
        absent @ (QueueReception::Empty | QueueReception::Closed) => Err(Failure::Reply(absent)),
    }
}

pub fn run() -> Result<(), Failure> {
    run_with_writer(&mut io::stdout().lock())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_real_mio_wake_conducts_the_same_program() -> Result<(), Failure> {
        let mut bytes = Vec::new();
        run_with_writer(&mut bytes)?;
        assert_eq!(bytes, b"Activated\n");
        Ok(())
    }

    #[test]
    fn an_output_failure_is_not_success() {
        struct Refuse;
        impl Write for Refuse {
            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
                Err(io::ErrorKind::BrokenPipe.into())
            }
            fn flush(&mut self) -> io::Result<()> {
                Ok(())
            }
        }
        assert!(
            matches!(run_with_writer(&mut Refuse), Err(Failure::Output { reply: Activation::Activated, failure: error }) if error.kind() == io::ErrorKind::BrokenPipe)
        );
    }
}