use core::any::type_name;
use core::mem::size_of;
use crate::{FlowObservation, FlowObserver, NoopFlowObserver, Pipe, 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, Pipe, StateMachine};
pub(super) trait TurnLaw<'message> {}
impl<'message, Machine> TurnLaw<'message> for Machine
where
Machine: StateMachine + Pipe,
Machine::In<'message>: CircuitInputOfDraft<Machine, Outcome = Machine::Out<'message>>,
{
}
}
#[doc(hidden)]
#[allow(private_bounds)]
pub trait TurnsCircuitDraftInputToOutput<'message>:
StateMachine + Pipe + 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 + Pipe,
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 + Pipe + 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 Pipe 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);
}
}