use core::convert::Infallible;
use pipecircuit::{
DriveFailure, Inline, InstalledPipe, Pipe, PipeDriver, PipeOutputHandler, PipeSender,
ReceiveFromOutcome, ScopedSender, StateMachine, StepInputOf,
};
pub struct Bytes;
impl Pipe for Bytes {
type In<'message> = &'message mut [u8];
type Out<'message> = &'message mut [u8];
}
pub struct BorrowingMachine<'state>(&'state mut usize);
impl StateMachine for BorrowingMachine<'_> {}
impl<'message> StepInputOf<BorrowingMachine<'_>> for &'message mut [u8] {
type Outcome = &'message mut [u8];
fn step(self, state: &mut BorrowingMachine<'_>) -> Self::Outcome {
*state.0 = self.len();
self
}
}
#[test]
fn inline_driver_preserves_stack_state_and_borrowed_payload_identity() {
let mut length = 0;
let mut bytes = *b"hello";
let address = bytes.as_ptr();
{
let mut pipe =
InstalledPipe::<Bytes, _>::new(Inline::for_pipe(BorrowingMachine(&mut length)));
let returned = pipe.outgoing().submit(&mut bytes);
assert_eq!(returned.as_ptr(), address);
assert_eq!(returned, b"hello");
}
assert_eq!(length, 5);
}
pub struct Source {
bytes: [u8; 4],
calls: usize,
}
pub struct Sender<'scope>(&'scope mut usize);
impl PipeSender<Bytes> for Sender<'_> {
type Outcome<'message> = usize;
fn send<'message>(&mut self, input: <Bytes as Pipe>::In<'message>) -> Self::Outcome<'message> {
*self.0 = input.len();
input.len()
}
}
impl<'message> ScopedSender<'message, Bytes> for Sender<'_> {
fn submit(&mut self, input: &'message mut [u8]) -> usize {
self.send(input)
}
}
impl PipeDriver<Bytes> for Source {
type Activation = ();
type Failure = Infallible;
type Submission<'message> = usize;
type Outgoing<'transport> = Sender<'transport>;
fn outgoing(&mut self) -> Self::Outgoing<'_> {
Sender(&mut self.calls)
}
fn drive<Handler>(
&mut self,
(): (),
handler: &mut Handler,
) -> ReceiveFromOutcome<(), (), DriveFailure<Infallible, Handler::Failure>>
where
Handler: PipeOutputHandler<Bytes, Self>,
{
let mut sender = Sender(&mut self.calls);
match handler.output(&mut self.bytes, &mut sender) {
Ok(()) => ReceiveFromOutcome::Progressed,
Err(failure) => ReceiveFromOutcome::Failed {
failure: DriveFailure::Conduct(failure),
},
}
}
}
pub struct Reply;
impl PipeOutputHandler<Bytes, Source> for Reply {
type Failure = Infallible;
fn output<'message, Outgoing>(
&mut self,
output: &'message mut [u8],
outgoing: &mut Outgoing,
) -> Result<(), Infallible>
where
Outgoing:
ScopedSender<'message, Bytes> + for<'input> PipeSender<Bytes, Outcome<'input> = usize>,
{
output.fill(7);
assert_eq!(outgoing.submit(output), 4);
Ok(())
}
}
#[test]
fn installed_driver_lends_receive_storage_and_its_sending_half_together() {
let mut pipe = InstalledPipe::<Bytes, _>::new(Source {
bytes: [0; 4],
calls: 0,
});
assert_eq!(pipe.drive((), &mut Reply), ReceiveFromOutcome::Progressed);
assert_eq!(pipe.transport().bytes, [7; 4]);
assert_eq!(pipe.transport().calls, 4);
}
#[test]
fn a_handler_cannot_return_driver_owned_receive_storage() {
let tests = trybuild::TestCases::new();
tests.compile_fail("tests/ui/driver_output_cannot_escape.rs");
tests.compile_fail("tests/ui/admitted_driver_requires_gate.rs");
}
#[cfg(feature = "programs")]
mod roster_source {
use super::*;
use pipecircuit::{CircuitFlowOutcome, CircuitSendOutcome, Switchboard, flow};
impl pipecircuit::AdmittedPipeDriver<Bytes> for Source {
type IntakeContext = usize;
fn drive_admitted<Handler>(
&mut self,
activation: (),
handler: &mut Handler,
) -> ReceiveFromOutcome<(), (), DriveFailure<Infallible, Handler::Failure>>
where
Handler:
PipeOutputHandler<Bytes, Self> + pipecircuit::IntakeGate<usize, Handler::Failure>,
{
match handler.admit(self.bytes.len()) {
Ok(pipecircuit::IntakeAdmission::Ready) => self.drive(activation, handler),
Ok(pipecircuit::IntakeAdmission::Paused) => ReceiveFromOutcome::Backpressured {
rejected: activation,
},
Err(failure) => ReceiveFromOutcome::Failed {
failure: DriveFailure::Conduct(failure),
},
}
}
}
pipecircuit::pipe_roster! {
mod program {
source: super::Bytes,
worker: super::Bytes,
}
}
struct Board {
address: *const u8,
}
impl Switchboard for Board {
type From<'message> = program::From<'message>;
type To<'message> = program::To<'message>;
fn route<'message>(&mut self, from: Self::From<'message>) -> Self::To<'message> {
match from {
program::From::Source(bytes) => {
assert_eq!(bytes.as_ptr(), self.address);
program::To::Worker(bytes)
}
program::From::Worker(bytes) => {
assert_eq!(bytes.as_ptr(), self.address);
program::To::Source(bytes)
}
}
}
}
struct Handler<'turn, 'state> {
board: Board,
peers: program::SourcePeers<
'turn,
InstalledPipe<Bytes, Inline<BorrowingMachine<'state>, Bytes>>,
>,
}
impl PipeOutputHandler<Bytes, Source> for Handler<'_, '_> {
type Failure = Infallible;
fn output<'message, Sender>(
&mut self,
output: &'message mut [u8],
outgoing: &mut Sender,
) -> Result<(), Infallible>
where
Sender: ScopedSender<'message, Bytes>
+ for<'input> PipeSender<Bytes, Outcome<'input> = usize>,
{
let outcome = self
.peers
.with_output(output, outgoing, |from, mut senders| {
flow(&mut self.board, from, |to| {
Ok::<_, Infallible>(match senders.submit(to) {
program::Submission::Worker(bytes) => CircuitSendOutcome::Immediate {
from: program::From::Worker(bytes),
},
program::Submission::Source(length) => {
CircuitSendOutcome::Completed { completion: length }
}
})
})
})?;
assert_eq!(outcome, CircuitFlowOutcome::Completed { completion: 4 });
Ok(())
}
}
#[test]
fn driver_storage_flows_through_a_borrowing_peer_and_back_to_its_source() {
let mut length = 0;
let mut pipes = program::Pipes {
source: InstalledPipe::new(Source {
bytes: [7; 4],
calls: 0,
}),
worker: InstalledPipe::new(Inline::for_pipe::<Bytes>(BorrowingMachine(&mut length))),
};
let address = pipes.source.transport().bytes.as_ptr();
let (source, peers) = pipes.borrow_pipes().split_source();
assert_eq!(
source.drive(
(),
&mut Handler {
board: Board { address },
peers
}
),
ReceiveFromOutcome::Progressed
);
assert_eq!(source.transport().calls, 4);
assert_eq!(source.transport().bytes, [7; 4]);
assert_eq!(length, 4);
}
#[test]
fn fallible_admission_forwards_outputs_and_preserves_conduction_failures() {
use pipecircuit::{
AdmissionFailure, AdmissionHandler, InlineSubmission, IntakeAdmission, SubmitWith,
};
let reason = String::from("submission refused after admission");
for refuse in [false, true] {
let mut length = 0;
let mut pipes = program::Pipes {
source: InstalledPipe::new(Source {
bytes: [7; 4],
calls: 0,
}),
worker: InstalledPipe::new(Inline::for_pipe::<Bytes>(BorrowingMachine(
&mut length,
))),
};
let mut board = Board {
address: pipes.source.transport().bytes.as_ptr(),
};
let mut circulation = program::Circulation::new(
&mut board,
program::SubmissionPolicies {
source: SubmitWith(|count: usize| {
assert_eq!(count, 4);
if refuse { Err(reason.as_str()) } else { Ok(()) }
}),
worker: InlineSubmission,
},
);
let (source, peers) = pipes.borrow_pipes().split_source();
let result = source.drive_admitted(
(),
&mut peers.handler(&mut circulation).with_fallible_admission(
|_, _, byte_count: usize| {
assert_eq!(byte_count, 4);
Ok::<_, u8>(IntakeAdmission::Ready)
},
),
);
if refuse {
assert!(matches!(result, ReceiveFromOutcome::Failed {
failure: DriveFailure::Conduct(AdmissionFailure::Conduct(program::SubmissionFailure::Source(failure)))
} if core::ptr::eq(failure, reason.as_str())));
} else {
assert!(matches!(result, ReceiveFromOutcome::Progressed));
}
assert_eq!(source.transport().calls, 4);
assert_eq!(source.transport().bytes, [7; 4]);
assert_eq!(length, 4);
}
}
}
#[test]
fn addressing_physical_failures_preserves_owned_and_borrowed_conduction_failures()
-> Result<(), &'static str> {
let physical = Box::new(3_u8);
let conduct = Box::new(7_u8);
let physical_address = core::ptr::from_ref(physical.as_ref());
let conduct_address = core::ptr::from_ref(conduct.as_ref());
let failure = DriveFailure::PhysicalAndConduct { physical, conduct }
.map_physical(|physical| ("client", physical));
let DriveFailure::PhysicalAndConduct {
physical: (source, physical),
conduct,
} = failure
else {
return Err("both failures must survive source addressing");
};
assert_eq!(source, "client");
assert_eq!(core::ptr::from_ref(physical.as_ref()), physical_address);
assert_eq!(core::ptr::from_ref(conduct.as_ref()), conduct_address);
let value = String::from("borrowed policy failure");
let mut calls = Vec::new();
let mapped = DriveFailure::<u8, _>::Conduct(value.as_str()).map_physical(|physical| {
calls.push(physical);
physical
});
assert!(calls.is_empty());
assert_eq!(mapped, DriveFailure::Conduct(value.as_str()));
let physical = DriveFailure::<u8, &str>::Physical(3).map_physical(|physical| {
calls.push(physical);
("npcs", physical)
});
assert_eq!(calls, [3]);
assert_eq!(physical, DriveFailure::Physical(("npcs", 3)));
Ok(())
}