pipecircuit 0.0.19

extern crate alloc;

use core::convert::Infallible;
use fibre::{
    TryRecvError, TrySendError,
    spsc::{self, BoundedSyncReceiver, BoundedSyncSender},
};
use pipecircuit::{
    Inline, InstalledPipe, Pipe, ReceiveFromOutcome, StateMachine, StepInputOf, Switchboard,
    TransportFrom, TransportTo,
};

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

#[derive(Debug, Eq, PartialEq)]
enum CounterEvent {
    Ready,
    Value(u8),
    Overflow { rejected: Add },
}

struct Counter(u8);
impl StateMachine for Counter {}
impl StepInputOf<Counter> for Add {
    type Outcome = CounterEvent;
    fn step(self, counter: &mut Counter) -> CounterEvent {
        match counter.0.checked_add(self.0) {
            Some(total) => {
                counter.0 = total;
                CounterEvent::Value(total)
            }
            None => CounterEvent::Overflow { rejected: self },
        }
    }
}

struct Counting;
impl Pipe for Counting {
    type In<'message> = Add;
    type Out<'message> = CounterEvent;
}

struct Program;
enum FromProgram {
    Client(Add),
    Counter(CounterEvent),
}
#[derive(Debug, Eq, PartialEq)]
enum ToProgram {
    Counter(Add),
    Client(CounterEvent),
}
impl Switchboard for Program {
    type From<'message> = FromProgram;
    type To<'message> = ToProgram;
    fn route<'message>(&mut self, from: Self::From<'message>) -> Self::To<'message> {
        match from {
            FromProgram::Client(input) => ToProgram::Counter(input),
            FromProgram::Counter(event) => ToProgram::Client(event),
        }
    }
}

struct QueueTransport {
    requests: BoundedSyncSender<Add>,
    events: BoundedSyncReceiver<CounterEvent>,
}
impl TransportTo<Counting> for QueueTransport {
    type Outcome<'transport, 'message> = Result<(), TrySendError<Add>>;
    fn send<'transport, 'message>(
        &'transport mut self,
        input: Add,
    ) -> Self::Outcome<'transport, 'message> {
        self.requests.try_send(input)
    }
}
impl TransportFrom<Counting> for QueueTransport {
    type IncomingMessage<'message> = ();
    type Failure = TryRecvError;
    fn receive_from(&mut self, (): ()) -> ReceiveFromOutcome<CounterEvent, (), TryRecvError> {
        match self.events.try_recv() {
            Ok(output) => ReceiveFromOutcome::Received { output },
            Err(TryRecvError::Empty) => ReceiveFromOutcome::Quiescent,
            Err(failure @ TryRecvError::Disconnected) => ReceiveFromOutcome::Failed { failure },
        }
    }
}

#[test]
fn one_contract_and_switchboard_accept_inline_and_foreign_program_implementations()
-> Result<(), &'static str> {
    let mut board = Program;
    let mut inline = InstalledPipe::<Counting, _>::new(Inline::for_pipe(Counter(0)));
    let ToProgram::Counter(input) = board.route(FromProgram::Client(Add(7))) else {
        return Err("request was not addressed to counter");
    };
    let inline_reply = board.route(FromProgram::Counter(inline.send(input)));
    assert_eq!(inline_reply, ToProgram::Client(CounterEvent::Value(7)));
    assert_eq!(inline.transport().destination().0, 7);

    let (requests, receiver) = spsc::bounded_sync(1);
    let (sender, events) = spsc::bounded_sync(2);
    let mut queued = InstalledPipe::<Counting, _>::new(QueueTransport { requests, events });
    let ToProgram::Counter(input) = board.route(FromProgram::Client(Add(7))) else {
        return Err("request was not addressed to counter");
    };
    assert_eq!(queued.send(input), Ok(()));
    assert_eq!(queued.receive_from(()), ReceiveFromOutcome::Quiescent);
    assert_eq!(queued.send(Add(9)), Err(TrySendError::Full(Add(9))));

    let peer = std::thread::spawn(move || {
        let mut counter = Counter(0);
        sender
            .try_send(CounterEvent::Ready)
            .map_err(|_failure| "ready event refused")?;
        let input = receiver
            .try_recv()
            .map_err(|_failure| "submitted input absent")?;
        sender
            .try_send(counter.step(input))
            .map_err(|_failure| "completion refused")?;
        Ok::<_, &'static str>(())
    });
    peer.join()
        .map_err(|_failure| "foreign program panicked")??;
    let ReceiveFromOutcome::Received { output } = queued.receive_from(()) else {
        return Err("ready event absent");
    };
    assert_eq!(
        board.route(FromProgram::Counter(output)),
        ToProgram::Client(CounterEvent::Ready)
    );
    let ReceiveFromOutcome::Received { output } = queued.receive_from(()) else {
        return Err("completion absent");
    };
    assert_eq!(board.route(FromProgram::Counter(output)), inline_reply);
    assert_eq!(
        queued.receive_from(()),
        ReceiveFromOutcome::Failed {
            failure: TryRecvError::Disconnected
        }
    );
    assert_eq!(queued.send(Add(11)), Err(TrySendError::Closed(Add(11))));
    Ok(())
}

struct Inspect;
impl Pipe for Inspect {
    type In<'message> = &'message str;
    type Out<'message> = &'message str;
}
struct Stateless;
impl StateMachine for Stateless {}
impl<'message> StepInputOf<Stateless> for &'message str {
    type Outcome = &'message str;
    fn step(self, _implementation: &mut Stateless) -> Self::Outcome {
        self.trim()
    }
}

#[test]
fn an_independent_inline_contract_preserves_borrowed_output() -> Result<(), Infallible> {
    let mut installed = InstalledPipe::<Inspect, _>::new(Inline::for_pipe(Stateless));
    let text = alloc::string::String::from("  borrowed  ");
    let output = installed.send(&text);
    assert_eq!(output, "borrowed");
    assert_eq!(output.as_ptr(), text.trim().as_ptr());
    Ok(())
}