use core::any::type_name;
use core::mem::size_of;
use crate::{
FlowObservation, FlowObserver, InstalledPipe, Pipe, StateMachine, StepInputOf, Switchboard,
TransportTo,
};
pub struct PipeFlow<Observer> {
observer: Observer,
}
impl<Observer> PipeFlow<Observer> {
pub const fn new(observer: Observer) -> Self {
Self { observer }
}
pub const fn observer(&self) -> &Observer {
&self.observer
}
pub const fn observer_mut(&mut self) -> &mut Observer {
&mut self.observer
}
pub fn into_observer(self) -> Observer {
self.observer
}
}
impl<Observer: FlowObserver> PipeFlow<Observer> {
#[must_use = "pipe outcomes must be routed or settled"]
#[inline(always)]
pub fn send_through<'transport, 'message, Destination, InstalledTransport>(
&mut self,
pipe: &'transport mut InstalledPipe<Destination, InstalledTransport>,
input: Destination::In<'message>,
) -> InstalledTransport::Outcome<'transport, 'message>
where
Destination: Pipe,
InstalledTransport: TransportTo<Destination>,
{
self.observer.observe(FlowObservation::PipeSendBegan {
participant: type_name::<Destination>(),
input: type_name::<Destination::In<'message>>(),
transport: type_name::<InstalledTransport>(),
input_representation_bytes: size_of::<Destination::In<'message>>(),
});
let outcome = pipe.send(input);
self.observer.observe(FlowObservation::PipeSendCompleted {
participant: type_name::<Destination>(),
transport: type_name::<InstalledTransport>(),
outcome: type_name::<InstalledTransport::Outcome<'transport, 'message>>(),
output_representation_bytes: size_of::<
InstalledTransport::Outcome<'transport, 'message>,
>(),
});
outcome
}
#[must_use = "participant step outcomes must be routed or settled"]
#[inline(always)]
pub fn step<Machine, Input>(&mut self, machine: &mut Machine, input: Input) -> Input::Outcome
where
Machine: StateMachine,
Input: StepInputOf<Machine>,
{
self.observer
.observe(FlowObservation::ParticipantStepBegan {
participant: type_name::<Machine>(),
input: type_name::<Input>(),
input_representation_bytes: size_of::<Input>(),
});
let outcome = machine.step(input);
self.observer
.observe(FlowObservation::ParticipantStepCompleted {
participant: type_name::<Machine>(),
outcome: type_name::<Input::Outcome>(),
output_representation_bytes: size_of::<Input::Outcome>(),
});
outcome
}
#[must_use = "switchboard route outcomes must be routed or settled"]
#[inline(always)]
pub fn route<'message, Board: Switchboard>(
&mut self,
switchboard: &mut Board,
input: Board::From<'message>,
) -> Board::To<'message> {
self.observer
.observe(FlowObservation::SwitchboardRouteBegan {
switchboard: type_name::<Board>(),
input: type_name::<Board::From<'message>>(),
input_representation_bytes: size_of::<Board::From<'message>>(),
});
let output = switchboard.route(input);
self.observer
.observe(FlowObservation::SwitchboardRouteCompleted {
switchboard: type_name::<Board>(),
output: type_name::<Board::To<'message>>(),
output_representation_bytes: size_of::<Board::To<'message>>(),
});
output
}
}