#![cfg(feature = "programs")]
use pipecircuit::{
Inline, InstalledPipe, Pipe, PipeRoster, StateMachine, StepInputOf, Switchboard,
};
struct Bytes;
impl Pipe for Bytes {
type In<'message> = &'message mut [u8];
type Out<'message> = &'message mut [u8];
}
pipecircuit::pipe_roster! {
mod program {
client: super::Bytes,
mirror: super::Bytes,
}
}
struct Board;
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>,
) -> pipecircuit::Route<Self::To<'message>> {
pipecircuit::Route::one({
match from {
program::From::Client(bytes) => program::To::Mirror(bytes),
program::From::Mirror(bytes) => program::To::Client(bytes),
}
})
}
}
struct Fill<'state>(&'state u8);
impl StateMachine for Fill<'_> {}
impl<'message> StepInputOf<Fill<'_>> for &'message mut [u8] {
type Outcome = &'message mut [u8];
fn step(self, machine: &mut Fill<'_>) -> Self::Outcome {
self.fill(*machine.0);
self
}
}
#[test]
fn one_roster_preserves_borrowed_routing_stack_state_and_distinct_same_protocol_roles()
-> Result<(), &'static str> {
let (first, second) = (3, 9);
let mut pipes = program::Pipes {
client: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&first))),
mirror: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&second))),
};
let mut payload = [0; 4];
let address = payload.as_ptr();
let output =
{
let (client, mut peers) = pipes.borrow_pipes().split_client();
peers.with_output(&mut payload, &mut client.outgoing(), |from, mut senders| {
match senders.send(
Board
.route(from)
.into_single()
.map_err(|_| "expected one destination")?,
) {
program::Submission::Mirror(output) => Ok(output),
program::Submission::Client(_) => Err("wrong destination for client output"),
}
})?
};
assert_eq!(output.as_ptr(), address);
assert_eq!(output, &[9; 4]);
let output = {
let (mirror, mut peers) = pipes.borrow_pipes().split_mirror();
peers.with_output(
output,
&mut mirror.outgoing(),
|from, mut senders| match senders.submit(
Board
.route(from)
.into_single()
.map_err(|_| "expected one destination")?,
) {
program::Submission::Client(output) => Ok(output),
program::Submission::Mirror(_) => Err("wrong destination for mirror output"),
},
)?
};
assert_eq!(output.as_ptr(), address);
assert_eq!(output, &[3; 4]);
assert_eq!(program::Schema::PIPE_NAMES, &["client", "mirror"]);
Ok(())
}
#[test]
fn roster_contracts() {
let cases = trybuild::TestCases::new();
cases.compile_fail("tests/ui/roster/*_fail.rs");
cases.pass("tests/ui/roster/*_pass.rs");
}
#[test]
fn circulation_preserves_lending_outputs_and_captured_policy_state() -> Result<(), &'static str> {
use pipecircuit::{InlineSubmission, SubmitWith};
let (first, second) = (3, 9);
let mut pipes = program::Pipes {
client: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&first))),
mirror: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&second))),
};
let mut bytes = [0; 4];
let address = bytes.as_ptr();
let mut settled = 0;
let mut board = Board;
let mut circulation = program::Circulation::new(
&mut board,
program::SubmissionPolicies {
client: SubmitWith(|output: &mut [u8]| {
assert_eq!(output.as_ptr(), address);
assert_eq!(output, &[3; 4]);
assert_eq!(settled, 0);
settled = 1;
Ok::<(), &'static str>(())
}),
mirror: InlineSubmission,
},
);
let (client, mut peers) = pipes.borrow_pipes().split_client();
let result = peers.with_output(&mut bytes, &mut client.outgoing(), |from, senders| {
program::Conductor::<Inline<Fill<'_>, Bytes>, Inline<Fill<'_>, Bytes>>::conduct(
&mut circulation,
from,
senders,
)
});
assert!(result.is_ok());
assert_eq!(settled, 1);
assert_eq!(bytes, [3; 4]);
Ok(())
}
#[test]
fn circulation_preserves_borrowed_failures_and_nominal_role_identity() {
use pipecircuit::SubmitWith;
let value = 7;
let mut pipes = program::Pipes {
client: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&value))),
mirror: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&value))),
};
let reason = String::from("physical refusal");
let borrowed = reason.as_str();
let mut board = Board;
let mut circulation = program::Circulation::new(
&mut board,
program::SubmissionPolicies {
client: SubmitWith(|_: &mut [u8]| Err(borrowed)),
mirror: SubmitWith(|_: &mut [u8]| Err(borrowed)),
},
);
let mut bytes = [0; 2];
let (client, mut peers) = pipes.borrow_pipes().split_client();
let result = peers.with_output(&mut bytes, &mut client.outgoing(), |from, senders| {
program::Conductor::<Inline<Fill<'_>, Bytes>, Inline<Fill<'_>, Bytes>>::conduct(
&mut circulation,
from,
senders,
)
});
assert!(
matches!(result, Err(program::SubmissionFailure::Mirror(failure)) if core::ptr::eq(failure, borrowed))
);
let (mirror, mut peers) = pipes.borrow_pipes().split_mirror();
let result = peers.with_output(&mut bytes, &mut mirror.outgoing(), |from, senders| {
program::Conductor::<Inline<Fill<'_>, Bytes>, Inline<Fill<'_>, Bytes>>::conduct(
&mut circulation,
from,
senders,
)
});
assert!(
matches!(result, Err(program::SubmissionFailure::Client(failure)) if core::ptr::eq(failure, borrowed))
);
}
#[test]
fn borrowed_policy_record_survives_source_turns_and_refusals() -> Result<(), &'static str> {
use core::convert::Infallible;
use pipecircuit::{CircuitSendOutcome, PipeOutputHandler, SubmissionPolicy};
struct Identity(u8);
struct Policy<'reason> {
identity: Box<Identity>,
calls: u8,
refusal: &'reason str,
}
impl<'message, 'reason> SubmissionPolicy<&'message mut [u8], &'message mut [u8]>
for Policy<'reason>
{
type Failure = &'reason str;
fn settle(
&mut self,
_: &'message mut [u8],
) -> Result<CircuitSendOutcome<&'message mut [u8], Infallible>, Self::Failure> {
self.calls = self.calls.checked_add(1).ok_or("call overflow")?;
if self.calls == 2 {
Err(self.refusal)
} else {
Ok(CircuitSendOutcome::Submitted)
}
}
}
let (first, second) = (3, 9);
let mut pipes = program::Pipes {
client: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&first))),
mirror: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&second))),
};
let reason = String::from("explicit policy refusal");
let mut policies = program::SubmissionPolicies {
client: Policy {
identity: Box::new(Identity(7)),
calls: 0,
refusal: &reason,
},
mirror: Policy {
identity: Box::new(Identity(7)),
calls: 0,
refusal: &reason,
},
};
let client_identity = core::ptr::from_ref(policies.client.identity.as_ref());
let mirror_identity = core::ptr::from_ref(policies.mirror.identity.as_ref());
let mut board = Board;
for turn in 1..=3 {
let mut bytes = [0; 4];
let (pipe, peers) = pipes.borrow_pipes().split_client();
let result = PipeOutputHandler::<Bytes, Inline<Fill<'_>, Bytes>>::output(
&mut peers.handler(&mut policies.circulation(&mut board)),
&mut bytes,
&mut pipe.outgoing(),
);
if turn == 2 {
assert!(
matches!(result, Err(program::SubmissionFailure::Mirror(failure))
if core::ptr::eq(failure, reason.as_str()))
);
} else {
assert!(result.is_ok());
}
assert_eq!(bytes, [9; 4]);
assert_eq!(policies.mirror.calls, turn);
let (pipe, peers) = pipes.borrow_pipes().split_mirror();
let result = PipeOutputHandler::<Bytes, Inline<Fill<'_>, Bytes>>::output(
&mut peers.handler(&mut policies.circulation(&mut board)),
&mut bytes,
&mut pipe.outgoing(),
);
if turn == 2 {
assert!(
matches!(result, Err(program::SubmissionFailure::Client(failure))
if core::ptr::eq(failure, reason.as_str()))
);
} else {
assert!(result.is_ok());
}
assert_eq!(bytes, [3; 4]);
assert_eq!(policies.client.calls, turn);
assert_eq!(policies.client.identity.0, 7);
assert_eq!(policies.mirror.identity.0, 7);
assert_eq!(
core::ptr::from_ref(policies.client.identity.as_ref()),
client_identity
);
assert_eq!(
core::ptr::from_ref(policies.mirror.identity.as_ref()),
mirror_identity
);
}
Ok(())
}
struct Fork;
impl Switchboard for Fork {
type From<'message> = program::From<'message>;
type To<'message> = program::To<'message>;
fn route<'message>(
&mut self,
from: Self::From<'message>,
) -> pipecircuit::Route<Self::To<'message>> {
match from {
program::From::Client(bytes) | program::From::Mirror(bytes) => {
let (first, second) = bytes.split_at_mut(bytes.len() / 2);
pipecircuit::Route::fanout((
program::To::Client(first),
program::To::Mirror(second),
))
}
}
}
}
#[test]
fn generated_circulation_conducts_fanout_only_with_its_declared_capacity() {
use pipecircuit::SubmitWith;
let (first, second) = (3, 9);
let mut pipes = program::Pipes {
client: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&first))),
mirror: InstalledPipe::new(Inline::for_pipe::<Bytes>(Fill(&second))),
};
let mut board = Fork;
let policies = || program::SubmissionPolicies {
client: SubmitWith(|_: &mut [u8]| Ok::<(), core::convert::Infallible>(())),
mirror: SubmitWith(|_: &mut [u8]| Ok::<(), core::convert::Infallible>(())),
};
let mut bytes = [0; 4];
{
let mut policies = policies();
let mut circulation = policies.circulation(&mut board);
let (client, mut peers) = pipes.borrow_pipes().split_client();
let result = peers.with_output(&mut bytes, &mut client.outgoing(), |from, senders| {
program::Conductor::<Inline<Fill<'_>, Bytes>, Inline<Fill<'_>, Bytes>>::conduct(
&mut circulation,
from,
senders,
)
});
assert!(matches!(
result,
Err(program::SubmissionFailure::RoutingCapacity {
maximum_pending: 0,
pending_destinations: 0,
refused_destinations: 2
})
));
}
assert_eq!(bytes, [0; 4]);
{
let mut policies = policies();
let mut circulation = policies.circulation(&mut board).with_fanout_capacity::<1>();
let (client, mut peers) = pipes.borrow_pipes().split_client();
let result = peers.with_output(&mut bytes, &mut client.outgoing(), |from, senders| {
program::Conductor::<Inline<Fill<'_>, Bytes>, Inline<Fill<'_>, Bytes>>::conduct(
&mut circulation,
from,
senders,
)
});
assert!(result.is_ok());
}
assert_eq!(bytes, [3, 3, 9, 9]);
}