pipecircuit-cli 0.0.5

use core::convert::Infallible;
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, From, To};
use pipecircuit::{
    Circuit, CircuitFlowOutcome, CircuitReactionOutcome, CircuitSendOutcome, ConductionFailure,
    Inline, InstalledPipe, Pipecircuit, Reactor, ReceiveFromOutcome, TransportFrom, TransportTo,
    flow,
};

// This finite starter has one physical startup event and exits after reporting.
// Long-running applications install a real reactor and its supported transports.
#[derive(Debug, Eq, PartialEq)]
pub struct StartupExhausted;

struct Startup {
    pending: bool,
}
impl Reactor for Startup {
    type Reaction = ();
    type Failure = StartupExhausted;

    fn wait_for_reaction(&mut self) -> Result<(), StartupExhausted> {
        if core::mem::replace(&mut self.pending, false) {
            Ok(())
        } else {
            Err(StartupExhausted)
        }
    }
}

struct ClientTransport<Writer>(Writer);
impl<Writer> TransportFrom<Client> for ClientTransport<Writer> {
    type IncomingMessage<'message>
        = ()
    where
        Self: 'message;
    type Failure = Infallible;

    fn receive_from(&mut self, (): ()) -> ReceiveFromOutcome<Activate, (), Infallible> {
        ReceiveFromOutcome::Received { output: Activate }
    }
}

impl<Writer: Write> TransportTo<Client> for ClientTransport<Writer> {
    type Outcome<'transport, 'message>
        = io::Result<()>
    where
        Self: 'transport;

    fn send<'transport, 'message>(
        &'transport mut self,
        input: Activation,
    ) -> Self::Outcome<'transport, 'message> {
        writeln!(self.0, "{input:?}")
    }
}

struct Installation<Writer> {
    reactor: Startup,
    client: InstalledPipe<Client, ClientTransport<Writer>>,
    activator: InstalledPipe<Activator, Inline<Latch, Activator>>,
}

impl<Writer: Write> Circuit<ActivationProgram> for Installation<Writer> {
    type Reactor = Startup;
    type Completion = ();
    type Failure = io::Error;

    fn reactor(&mut self) -> &mut Startup {
        &mut self.reactor
    }

    fn react(
        &mut self,
        board: &mut ActivationProgram,
        (): &(),
    ) -> Result<CircuitReactionOutcome<()>, io::Error> {
        let output = match self.client.receive_from(()) {
            ReceiveFromOutcome::Received { output } => output,
            ReceiveFromOutcome::Progressed => return Ok(CircuitReactionOutcome::ContinueReaction),
            ReceiveFromOutcome::Quiescent | ReceiveFromOutcome::Backpressured { rejected: () } => {
                return Ok(CircuitReactionOutcome::AwaitingReaction);
            }
            ReceiveFromOutcome::Failed { failure } => match failure {},
        };
        let result = flow(board, From::Client(output), |to| -> Result<_, io::Error> {
            match to {
                To::Activator(input) => Ok(CircuitSendOutcome::Immediate {
                    from: From::Activator(self.activator.send(input)),
                }),
                To::Client(input) => {
                    self.client.send(input)?;
                    Ok(CircuitSendOutcome::Completed { completion: () })
                }
            }
        })?;
        match result {
            CircuitFlowOutcome::Submitted => Ok(CircuitReactionOutcome::ContinueReaction),
            CircuitFlowOutcome::Completed { completion } => {
                Ok(CircuitReactionOutcome::Completed { completion })
            }
        }
    }
}

fn install<Writer: Write>(writer: Writer) -> Pipecircuit<ActivationProgram, Installation<Writer>> {
    Pipecircuit::new(
        ActivationProgram,
        Installation {
            reactor: Startup { pending: true },
            client: InstalledPipe::new(ClientTransport(writer)),
            activator: InstalledPipe::new(Inline::for_pipe(Latch::default())),
        },
    )
}

pub fn run() -> Result<(), ConductionFailure<StartupExhausted, io::Error>> {
    let mut pipecircuit = install(io::stdout().lock());
    pipecircuit.conduct()
}

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

    #[test]
    fn startup_is_consumed_once() {
        let mut reactor = Startup { pending: true };
        assert_eq!(reactor.wait_for_reaction(), Ok(()));
        assert_eq!(reactor.wait_for_reaction(), Err(StartupExhausted));
    }

    #[test]
    fn conducts_the_program_and_reports_once() {
        let mut bytes = Vec::new();
        assert!(install(&mut bytes).conduct().is_ok());
        assert_eq!(bytes, b"Activated\n");
    }

    #[test]
    fn preserves_the_output_failure() {
        struct Refuse;
        impl Write for Refuse {
            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
                Err(io::Error::from(io::ErrorKind::BrokenPipe))
            }
            fn flush(&mut self) -> io::Result<()> {
                Ok(())
            }
        }
        let outcome = install(Refuse).conduct();
        assert!(
            matches!(outcome, Err(ConductionFailure::Circuit { failure: error }) if error.kind() == io::ErrorKind::BrokenPipe)
        );
    }
}