pipecircuit 0.0.23

use core::convert::Infallible;

use pipecircuit::{
    Circuit, CircuitFlowOutcome, CircuitReactionOutcome, CircuitSendOutcome, Inline, InstalledPipe,
    NoInput, NoOutput, Pipe, Pipecircuit, Reactor, ReceiveFromOutcome, StateMachine, StepInputOf,
    Switchboard, TransportFrom, TransportTo, flow,
};

struct Startup;

impl Pipe for Startup {
    type In<'message> = NoInput;
    type Out<'message> = &'message str;
}

#[derive(Clone, Copy)]
struct StartupReady;

struct StartupReactor;

impl Reactor for StartupReactor {
    type Reaction = StartupReady;
    type Failure = Infallible;

    fn wait_for_reaction(&mut self) -> Result<Self::Reaction, Self::Failure> {
        Ok(StartupReady)
    }
}

struct StartupTransport {
    output: &'static str,
    received_messages: u8,
}

impl TransportFrom<Startup> for StartupTransport {
    type IncomingMessage<'message> = StartupReady;
    type Failure = Infallible;

    fn receive_from<'message>(
        &'message mut self,
        _incoming_message: Self::IncomingMessage<'message>,
    ) -> ReceiveFromOutcome<
        <Startup as Pipe>::Out<'message>,
        Self::IncomingMessage<'message>,
        Self::Failure,
    > {
        self.received_messages = self.received_messages.saturating_add(1);
        ReceiveFromOutcome::Received {
            output: self.output,
        }
    }
}

struct Accumulator(u16);

impl StateMachine for Accumulator {}

impl Pipe for Accumulator {
    type In<'message> = Add<'message>;
    type Out<'message> = Total;
}

struct Add<'message>(&'message str);
struct Total(u16);

impl StepInputOf<Accumulator> for Add<'_> {
    type Outcome = Total;

    fn step(self, accumulator: &mut Accumulator) -> Self::Outcome {
        let addition = match self.0 {
            "1" => 1,
            "7" => 7,
            _ => 0,
        };
        accumulator.0 = accumulator.0.saturating_add(addition);
        Total(accumulator.0)
    }
}

struct Exit;
struct ExitWith(u16);

impl Pipe for Exit {
    type In<'message> = ExitWith;
    type Out<'message> = NoOutput;
}

struct ExitTransport;

impl TransportTo<Exit> for ExitTransport {
    type Outcome<'transport, 'message> = u16;

    fn send<'transport, 'message>(
        &'transport mut self,
        input: ExitWith,
    ) -> Self::Outcome<'transport, 'message> {
        input.0
    }
}

enum ProgramIn<'message> {
    Startup(<Startup as Pipe>::Out<'message>),
    Accumulator(Total),
}

enum ProgramOut<'message> {
    Accumulator(Add<'message>),
    Exit(ExitWith),
}

struct ProgramSwitchboard {
    routed_messages: u8,
}

impl Switchboard for ProgramSwitchboard {
    type From<'message> = ProgramIn<'message>;
    type To<'message> = ProgramOut<'message>;

    fn route<'message>(
        &mut self,
        input: Self::From<'message>,
    ) -> pipecircuit::Route<Self::To<'message>> {
        pipecircuit::Route::one({
            self.routed_messages = self.routed_messages.saturating_add(1);
            match input {
                ProgramIn::Startup(value) => ProgramOut::Accumulator(Add(value)),
                ProgramIn::Accumulator(Total(total)) if total < 10 => {
                    ProgramOut::Accumulator(Add("1"))
                }
                ProgramIn::Accumulator(Total(total)) => ProgramOut::Exit(ExitWith(total)),
            }
        })
    }
}

struct ProgramCircuit {
    reactor: StartupReactor,
    startup_pipe: InstalledPipe<Startup, StartupTransport>,
    accumulator_pipe: InstalledPipe<Accumulator, Inline<Accumulator>>,
    exit_pipe: InstalledPipe<Exit, ExitTransport>,
}

impl ProgramCircuit {
    fn send<'message>(
        accumulator_pipe: &mut InstalledPipe<Accumulator, Inline<Accumulator>>,
        exit_pipe: &mut InstalledPipe<Exit, ExitTransport>,
        to: ProgramOut<'message>,
    ) -> Result<CircuitSendOutcome<ProgramIn<'message>, ProgramConductionOutcome>, Infallible> {
        match to {
            ProgramOut::Accumulator(input) => Ok(CircuitSendOutcome::Immediate {
                from: ProgramIn::Accumulator(accumulator_pipe.send(input)),
            }),
            ProgramOut::Exit(input) => Ok(CircuitSendOutcome::Completed {
                completion: ProgramConductionOutcome::Exited {
                    total: exit_pipe.send(input),
                },
            }),
        }
    }
}

#[derive(Debug, Eq, PartialEq)]
enum ProgramConductionOutcome {
    Exited { total: u16 },
    StartupQuiescent,
}

impl Circuit<ProgramSwitchboard> for ProgramCircuit {
    type Reactor = StartupReactor;
    type Completion = ProgramConductionOutcome;
    type Failure = &'static str;

    fn reactor(&mut self) -> &mut Self::Reactor {
        &mut self.reactor
    }

    fn react(
        &mut self,
        switchboard: &mut ProgramSwitchboard,
        reaction: &StartupReady,
    ) -> Result<CircuitReactionOutcome<Self::Completion>, Self::Failure> {
        let startup_output = match self.startup_pipe.receive_from(*reaction) {
            ReceiveFromOutcome::Received { output } => output,
            ReceiveFromOutcome::Progressed => {
                return Ok(CircuitReactionOutcome::ContinueReaction);
            }
            ReceiveFromOutcome::Quiescent => {
                return Ok(CircuitReactionOutcome::Completed {
                    completion: ProgramConductionOutcome::StartupQuiescent,
                });
            }
            ReceiveFromOutcome::Backpressured { rejected: _ } => {
                return Ok(CircuitReactionOutcome::AwaitingReaction);
            }
            ReceiveFromOutcome::Failed { failure } => match failure {},
        };
        match flow(
            switchboard,
            ProgramIn::Startup(startup_output),
            &mut pipecircuit::PendingRoutes::<_, 0>::new(),
            |to| Self::send(&mut self.accumulator_pipe, &mut self.exit_pipe, to),
        )
        .map_err(|_| "single-route capacity exceeded")?
        {
            CircuitFlowOutcome::Submitted => Ok(CircuitReactionOutcome::ContinueReaction),
            CircuitFlowOutcome::Completed { completion } => {
                Ok(CircuitReactionOutcome::Completed { completion })
            }
        }
    }
}

#[test]
fn one_switchboard_drives_one_installed_physical_circuit() {
    let circuit = ProgramCircuit {
        reactor: StartupReactor,
        startup_pipe: InstalledPipe::new(StartupTransport {
            output: "7",
            received_messages: 0,
        }),
        accumulator_pipe: InstalledPipe::new(Inline::new(Accumulator(0))),
        exit_pipe: InstalledPipe::new(ExitTransport),
    };
    let mut program = Pipecircuit::new(ProgramSwitchboard { routed_messages: 0 }, circuit);

    assert_eq!(
        program.conduct(),
        Ok(ProgramConductionOutcome::Exited { total: 10 })
    );
    assert_eq!(
        program.circuit().startup_pipe.transport().received_messages,
        1
    );
    assert_eq!(
        program
            .circuit()
            .accumulator_pipe
            .transport()
            .destination()
            .0,
        10
    );
    assert_eq!(program.switchboard().routed_messages, 5);
}