pipecircuit 0.0.23

use core::marker::PhantomData;

use crate::{DuplexTransport, ReceiveFromOutcome, TransportFrom, TransportTo};

/// A nominal duplex semantic contract, independent of its implementation and transport.
/// Submission does not imply a synchronous response; either direction may be uninhabited.
pub trait Pipe {
    type In<'message>;
    type Out<'message>;
}

pub enum NoInput {}

pub enum NoOutput {}

pub struct InstalledPipe<Protocol: Pipe, Transport> {
    transport: Transport,
    protocol: PhantomData<fn() -> Protocol>,
}

impl<Protocol: Pipe, Transport> InstalledPipe<Protocol, Transport> {
    pub const fn new(transport: Transport) -> Self {
        Self {
            transport,
            protocol: PhantomData,
        }
    }

    pub const fn transport(&self) -> &Transport {
        &self.transport
    }

    pub const fn transport_mut(&mut self) -> &mut Transport {
        &mut self.transport
    }

    pub fn into_transport(self) -> Transport {
        self.transport
    }
}

impl<Protocol, Transport> InstalledPipe<Protocol, Transport>
where
    Protocol: Pipe,
    Transport: crate::PipeDriver<Protocol>,
{
    pub fn outgoing(&mut self) -> Transport::Outgoing<'_> {
        self.transport.outgoing()
    }

    pub fn drive<Handler>(
        &mut self,
        activation: Transport::Activation,
        handler: &mut Handler,
    ) -> ReceiveFromOutcome<
        (),
        Transport::Activation,
        crate::DriveFailure<Transport::Failure, Handler::Failure>,
    >
    where
        Handler: crate::PipeOutputHandler<Protocol, Transport>,
    {
        self.transport.drive(activation, handler)
    }
}

impl<Protocol, Transport> InstalledPipe<Protocol, Transport>
where
    Protocol: Pipe,
    Transport: crate::AdmittedPipeDriver<Protocol>,
{
    pub fn drive_admitted<Handler>(
        &mut self,
        activation: Transport::Activation,
        handler: &mut Handler,
    ) -> ReceiveFromOutcome<
        (),
        Transport::Activation,
        crate::DriveFailure<Transport::Failure, Handler::Failure>,
    >
    where
        Handler: crate::PipeOutputHandler<Protocol, Transport>
            + crate::IntakeGate<Transport::IntakeContext, Handler::Failure>,
    {
        self.transport.drive_admitted(activation, handler)
    }
}

impl<Protocol, Transport> InstalledPipe<Protocol, Transport>
where
    Protocol: Pipe,
    Transport: DuplexTransport<Protocol>,
{
    pub fn split(&mut self) -> (Transport::Incoming<'_>, Transport::Outgoing<'_>) {
        self.transport.split()
    }
}

impl<Protocol, Transport> InstalledPipe<Protocol, Transport>
where
    Protocol: Pipe,
    Transport: TransportTo<Protocol>,
{
    #[must_use = "pipe outcomes must be routed or settled"]
    pub fn send<'transport, 'message>(
        &'transport mut self,
        input: Protocol::In<'message>,
    ) -> Transport::Outcome<'transport, 'message> {
        self.transport.send(input)
    }
}

impl<Protocol, Transport> InstalledPipe<Protocol, Transport>
where
    Protocol: Pipe,
    Transport: TransportFrom<Protocol>,
{
    pub fn receive_from<'message>(
        &'message mut self,
        incoming_message: Transport::IncomingMessage<'message>,
    ) -> ReceiveFromOutcome<
        Protocol::Out<'message>,
        Transport::IncomingMessage<'message>,
        Transport::Failure,
    > {
        self.transport.receive_from(incoming_message)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Inline, StateMachine, StepInputOf};

    struct ActiveParticipant;

    impl crate::Pipe for ActiveParticipant {
        type In<'message> = &'message str;
        type Out<'message> = &'message str;
    }

    #[derive(Debug, Eq, PartialEq)]
    enum SubmissionOutcome {
        Submitted,
    }

    #[derive(Clone, Copy, Debug, Eq, PartialEq)]
    enum IncomingMessage<'message> {
        Output(&'message str),
        Stored,
        Progressed,
        Idle,
        Blocked(&'message str),
        Broken(u8),
    }

    struct ActiveTransport {
        submission_count: usize,
        received: [u8; 6],
    }

    impl TransportTo<ActiveParticipant> for ActiveTransport {
        type Outcome<'transport, 'message>
            = SubmissionOutcome
        where
            Self: 'transport;

        fn send<'transport, 'message>(
            &'transport mut self,
            _input: <ActiveParticipant as crate::Pipe>::In<'message>,
        ) -> Self::Outcome<'transport, 'message> {
            self.submission_count = self.submission_count.saturating_add(1);
            SubmissionOutcome::Submitted
        }
    }

    impl TransportFrom<ActiveParticipant> for ActiveTransport {
        type IncomingMessage<'message>
            = IncomingMessage<'message>
        where
            Self: 'message;
        type Failure = u8;

        fn receive_from<'message>(
            &'message mut self,
            incoming_message: Self::IncomingMessage<'message>,
        ) -> ReceiveFromOutcome<
            <ActiveParticipant as crate::Pipe>::Out<'message>,
            Self::IncomingMessage<'message>,
            Self::Failure,
        > {
            match incoming_message {
                IncomingMessage::Output(output) => ReceiveFromOutcome::Received { output },
                IncomingMessage::Stored => match core::str::from_utf8(&self.received) {
                    Ok(output) => ReceiveFromOutcome::Received { output },
                    Err(_) => ReceiveFromOutcome::Failed { failure: 0 },
                },
                IncomingMessage::Progressed => ReceiveFromOutcome::Progressed,
                IncomingMessage::Idle => ReceiveFromOutcome::Quiescent,
                IncomingMessage::Blocked(rejected) => ReceiveFromOutcome::Backpressured {
                    rejected: IncomingMessage::Blocked(rejected),
                },
                IncomingMessage::Broken(failure) => ReceiveFromOutcome::Failed { failure },
            }
        }
    }

    struct Recorder(usize);

    impl StateMachine for Recorder {}

    struct Record<'message>(&'message str);

    impl<'message> StepInputOf<Recorder> for Record<'message> {
        type Outcome = &'message str;

        fn step(self, recorder: &mut Recorder) -> Self::Outcome {
            recorder.0 = recorder.0.saturating_add(1);
            self.0
        }
    }

    struct Count;

    impl StepInputOf<Recorder> for Count {
        type Outcome = usize;

        fn step(self, recorder: &mut Recorder) -> Self::Outcome {
            recorder.0
        }
    }

    enum RecorderIn<'message> {
        Record(Record<'message>),
        Count(Count),
    }

    enum RecorderOut<'message> {
        Recorded(&'message str),
        Counted(usize),
    }

    impl crate::Pipe for Recorder {
        type In<'message> = RecorderIn<'message>;
        type Out<'message> = RecorderOut<'message>;
    }

    impl<'message> StepInputOf<Recorder> for RecorderIn<'message> {
        type Outcome = RecorderOut<'message>;

        fn step(self, recorder: &mut Recorder) -> Self::Outcome {
            match self {
                Self::Record(input) => RecorderOut::Recorded(input.step(recorder)),
                Self::Count(input) => RecorderOut::Counted(input.step(recorder)),
            }
        }
    }

    #[test]
    fn a_pipe_preserves_each_precise_destination_and_transport_outcome() {
        let mut pipe = InstalledPipe::<Recorder, _>::new(Inline::new(Recorder(0)));

        assert!(matches!(
            pipe.send(RecorderIn::Record(Record("borrowed"))),
            RecorderOut::Recorded("borrowed")
        ));
        assert!(matches!(
            pipe.send(RecorderIn::Count(Count)),
            RecorderOut::Counted(1)
        ));
        assert_eq!(pipe.transport().destination().0, 1);
        assert_eq!(pipe.transport_mut().destination().0, 1);
        assert_eq!(pipe.into_transport().into_destination().0, 1);
    }

    #[test]
    fn an_active_pipe_separates_submission_from_independent_output() {
        let mut pipe = InstalledPipe::<ActiveParticipant, _>::new(ActiveTransport {
            submission_count: 0,
            received: *b"stored",
        });

        assert_eq!(pipe.send("outbound"), SubmissionOutcome::Submitted);
        assert_eq!(
            pipe.receive_from(IncomingMessage::Output("inbound")),
            ReceiveFromOutcome::Received { output: "inbound" }
        );
        assert_eq!(
            pipe.receive_from(IncomingMessage::Stored),
            ReceiveFromOutcome::Received { output: "stored" }
        );
        assert_eq!(
            pipe.receive_from(IncomingMessage::Progressed),
            ReceiveFromOutcome::Progressed
        );
        assert_eq!(
            pipe.receive_from(IncomingMessage::Idle),
            ReceiveFromOutcome::Quiescent
        );
        assert_eq!(
            pipe.receive_from(IncomingMessage::Blocked("preserved")),
            ReceiveFromOutcome::Backpressured {
                rejected: IncomingMessage::Blocked("preserved")
            }
        );
        assert_eq!(
            pipe.receive_from(IncomingMessage::Broken(7)),
            ReceiveFromOutcome::Failed { failure: 7 }
        );
        assert_eq!(pipe.into_transport().submission_count, 1);
    }
}