pipecircuit 0.0.8

use core::any::type_name;
use core::mem::size_of;

use crate::{FlowObservation, FlowObserver, NoopFlowObserver, Participant, PipeFlow, StateMachine};

pub trait CircuitInputOfDraft<Machine: StateMachine> {
    type Outcome;

    #[must_use = "circuit input outcomes must be delivered or settled"]
    fn turn<Observer: FlowObserver>(
        self,
        circuit: &mut Machine,
        flow: &mut PipeFlow<Observer>,
    ) -> Self::Outcome;
}

mod private {
    use super::{CircuitInputOfDraft, Participant, StateMachine};

    pub(super) trait TurnLaw<'message> {}

    impl<'message, Machine> TurnLaw<'message> for Machine
    where
        Machine: StateMachine + Participant,
        Machine::In<'message>: CircuitInputOfDraft<Machine, Outcome = Machine::Out<'message>>,
    {
    }
}

#[doc(hidden)]
#[allow(private_bounds)]
pub trait TurnsCircuitDraftInputToOutput<'message>:
    StateMachine + Participant + private::TurnLaw<'message>
{
    #[must_use = "circuit turn outcomes must be delivered or settled"]
    #[doc(hidden)]
    fn turn_boundary<Observer: FlowObserver>(
        &mut self,
        input: Self::In<'message>,
        flow: &mut PipeFlow<Observer>,
    ) -> Self::Out<'message>;
}

impl<'message, Machine> TurnsCircuitDraftInputToOutput<'message> for Machine
where
    Machine: StateMachine + Participant,
    Machine::In<'message>: CircuitInputOfDraft<Machine, Outcome = Machine::Out<'message>>,
{
    #[inline(always)]
    fn turn_boundary<Observer: FlowObserver>(
        &mut self,
        input: Self::In<'message>,
        flow: &mut PipeFlow<Observer>,
    ) -> Self::Out<'message> {
        input.turn(self, flow)
    }
}

pub trait CircuitDraft:
    StateMachine + Participant + for<'message> TurnsCircuitDraftInputToOutput<'message>
{
    type Switchboard: crate::Switchboard;

    fn switchboard(&self) -> &Self::Switchboard;

    fn switchboard_mut(&mut self) -> &mut Self::Switchboard;

    #[must_use = "circuit turn outcomes must be delivered or settled"]
    #[inline(always)]
    fn turn<'message>(&mut self, input: Self::In<'message>) -> Self::Out<'message> {
        self.turn_in_flow(input, &mut PipeFlow::new(NoopFlowObserver))
    }

    #[must_use = "circuit turn outcomes must be delivered or settled"]
    #[doc(hidden)]
    #[inline(always)]
    fn turn_in_flow<'message, Observer: FlowObserver>(
        &mut self,
        input: Self::In<'message>,
        flow: &mut PipeFlow<Observer>,
    ) -> Self::Out<'message> {
        flow.observer_mut()
            .observe(FlowObservation::CircuitTurnBegan {
                circuit: type_name::<Self>(),
                input: type_name::<Self::In<'message>>(),
                input_representation_bytes: size_of::<Self::In<'message>>(),
            });
        let output =
            <Self as TurnsCircuitDraftInputToOutput<'message>>::turn_boundary(self, input, flow);
        flow.observer_mut()
            .observe(FlowObservation::CircuitTurnCompleted {
                circuit: type_name::<Self>(),
                output: type_name::<Self::Out<'message>>(),
                output_representation_bytes: size_of::<Self::Out<'message>>(),
            });
        output
    }

    fn with_observer<Observer: FlowObserver>(
        self,
        observer: Observer,
    ) -> ObservedCircuitDraft<Self, Observer> {
        ObservedCircuitDraft::new(self, observer)
    }
}

pub struct ObservedCircuitDraft<InstalledCircuit, Observer> {
    circuit: InstalledCircuit,
    observer: Observer,
}

impl<InstalledCircuit, Observer> ObservedCircuitDraft<InstalledCircuit, Observer> {
    pub const fn new(circuit: InstalledCircuit, observer: Observer) -> Self {
        Self { circuit, observer }
    }

    pub const fn circuit(&self) -> &InstalledCircuit {
        &self.circuit
    }

    pub const fn observer(&self) -> &Observer {
        &self.observer
    }

    pub const fn observer_mut(&mut self) -> &mut Observer {
        &mut self.observer
    }

    pub fn into_parts(self) -> (InstalledCircuit, Observer) {
        (self.circuit, self.observer)
    }
}

impl<InstalledCircuit: CircuitDraft, Observer: FlowObserver>
    ObservedCircuitDraft<InstalledCircuit, Observer>
{
    #[must_use = "observed circuit turn outcomes must be delivered or settled"]
    #[inline(always)]
    pub fn turn<'message>(
        &mut self,
        input: InstalledCircuit::In<'message>,
    ) -> InstalledCircuit::Out<'message> {
        self.circuit
            .turn_in_flow(input, &mut PipeFlow::new(&mut self.observer))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{CountingFlowObserver, ExecutionCounts, NoopFlowObserver, PipeFlow, Switchboard};

    #[derive(Debug, Eq, PartialEq)]
    struct TestSwitchboard;

    impl Switchboard for TestSwitchboard {
        type From<'message> = u8;
        type To<'message> = u16;

        fn route<'message>(&mut self, input: Self::From<'message>) -> Self::To<'message> {
            u16::from(input)
        }
    }

    #[derive(Debug, Eq, PartialEq)]
    struct TestCircuit {
        switchboard: TestSwitchboard,
    }

    impl StateMachine for TestCircuit {}

    impl Participant for TestCircuit {
        type In<'message> = u8;
        type Out<'message> = u16;
    }

    impl CircuitInputOfDraft<TestCircuit> for u8 {
        type Outcome = u16;

        fn turn<Observer: FlowObserver>(
            self,
            circuit: &mut TestCircuit,
            flow: &mut PipeFlow<Observer>,
        ) -> Self::Outcome {
            flow.route(&mut circuit.switchboard, self)
        }
    }

    impl CircuitDraft for TestCircuit {
        type Switchboard = TestSwitchboard;

        fn switchboard(&self) -> &Self::Switchboard {
            &self.switchboard
        }

        fn switchboard_mut(&mut self) -> &mut Self::Switchboard {
            &mut self.switchboard
        }
    }

    #[derive(Default)]
    struct CountingObserver(usize);

    impl FlowObserver for CountingObserver {
        fn observe(&mut self, _observation: FlowObservation) {
            self.0 = self.0.saturating_add(1);
        }
    }

    #[test]
    fn circuit_turns_use_their_declared_participant_boundary() {
        let mut circuit = TestCircuit {
            switchboard: TestSwitchboard,
        };

        assert_eq!(circuit.turn(7), 7);
        assert_eq!(circuit.switchboard(), &TestSwitchboard);
        assert_eq!(*circuit.switchboard_mut(), TestSwitchboard);
    }

    #[test]
    fn observed_circuits_reuse_one_observer_across_turns() {
        let mut program = TestCircuit {
            switchboard: TestSwitchboard,
        }
        .with_observer(CountingObserver::default());

        assert_eq!(program.turn(7), 7);
        assert_eq!(program.turn(8), 8);
        assert_eq!(
            program.circuit(),
            &TestCircuit {
                switchboard: TestSwitchboard,
            }
        );
        assert_eq!(program.observer().0, 8);
        assert_eq!(program.observer_mut().0, 8);
        let (_circuit, observer) = program.into_parts();
        assert_eq!(observer.0, 8);
    }

    #[test]
    fn counted_circuits_preserve_semantics_and_measure_representation_traffic() {
        let mut direct = TestCircuit {
            switchboard: TestSwitchboard,
        };
        let mut counted = TestCircuit {
            switchboard: TestSwitchboard,
        }
        .with_observer(CountingFlowObserver::new());

        assert_eq!(counted.turn(7), direct.turn(7));
        assert_eq!(counted.circuit(), &direct);

        let counts = counted.observer().counts();
        assert_eq!(counts.circuit_turns().started(), 1);
        assert_eq!(counts.circuit_turns().completed(), 1);
        assert_eq!(counts.circuit_turns().input_representation_bytes(), 1);
        assert_eq!(counts.circuit_turns().output_representation_bytes(), 2);
        assert_eq!(counts.switchboard_routes(), counts.circuit_turns());
        assert_eq!(counts.pipe_sends(), ExecutionCounts::new().pipe_sends());
        assert_eq!(
            counts.participant_steps(),
            ExecutionCounts::new().participant_steps()
        );
    }

    #[test]
    fn pipe_flow_accessors_preserve_the_observer() {
        let mut flow = PipeFlow::new(CountingObserver::default());

        assert_eq!(flow.observer().0, 0);
        assert_eq!(flow.observer_mut().0, 0);
        assert_eq!(flow.into_observer().0, 0);
        let _flow = PipeFlow::new(NoopFlowObserver);
    }
}