use core::convert::Infallible;
use core::str::Utf8Error;
use pipecircuit::{
Circuit, CircuitFlowOutcome, CircuitReactionOutcome, CircuitSendOutcome, EncodedByteCount,
Inline, InstalledPipe, MessageDecoder, MessageEncoder, NoInput, NoOutput, Pipe, Pipecircuit,
Reactor, ReceiveFromOutcome, StateMachine, StepInputOf, Switchboard, TransportFrom,
TransportTo, flow,
};
struct DecodedMessages;
impl Pipe for DecodedMessages {
type In<'message> = NoInput;
type Out<'message> = DecodedMessagesOut<'message>;
}
enum DecodedMessagesOut<'message> {
Message(&'message str),
Finished,
}
#[derive(Clone, Copy)]
struct PhysicalReaction(u8);
struct SequenceReactor {
next: u8,
waits: u8,
}
impl Reactor for SequenceReactor {
type Reaction = PhysicalReaction;
type Failure = Infallible;
fn wait_for_reaction(&mut self) -> Result<Self::Reaction, Self::Failure> {
self.waits = self.waits.saturating_add(1);
let reaction = PhysicalReaction(self.next);
self.next = self.next.saturating_add(1);
Ok(reaction)
}
}
struct Utf8MessageDecoder;
impl MessageDecoder for Utf8MessageDecoder {
type Message<'frame> = &'frame str;
type Error = Utf8Error;
fn decode<'frame>(
&self,
frame: &'frame mut [u8],
) -> Result<Self::Message<'frame>, Self::Error> {
core::str::from_utf8(frame)
}
}
#[derive(Debug)]
enum BorrowedBatchReceiveFailure {
RoutingCapacity,
ReceiveStorageFull,
MessageCapacityReached,
ReceiveByteUnavailable { byte: usize },
MessageRangeUnavailable { start: usize, end: usize },
InvalidUtf8 { source: Utf8Error },
}
impl core::fmt::Display for BorrowedBatchReceiveFailure {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::RoutingCapacity => formatter.write_str("routing capacity exceeded"),
Self::ReceiveStorageFull => formatter.write_str("receive storage is full"),
Self::MessageCapacityReached => formatter.write_str("message capacity was reached"),
Self::ReceiveByteUnavailable { byte } => {
write!(formatter, "receive byte {byte} is unavailable")
}
Self::MessageRangeUnavailable { start, end } => {
write!(formatter, "message range {start}..{end} is unavailable")
}
Self::InvalidUtf8 { source } => source.fmt(formatter),
}
}
}
struct BorrowedMessageBatchFitting {
decoder: Utf8MessageDecoder,
receive_storage: [u8; 64],
received_byte_count: usize,
next_scan_byte: usize,
next_message_byte: usize,
message_ranges: [(usize, usize); 8],
message_count: usize,
next_message: usize,
loaded_reaction: Option<u8>,
physical_reads: u8,
finished: bool,
}
impl BorrowedMessageBatchFitting {
const fn new() -> Self {
Self {
decoder: Utf8MessageDecoder,
receive_storage: [0; 64],
received_byte_count: 0,
next_scan_byte: 0,
next_message_byte: 0,
message_ranges: [(0, 0); 8],
message_count: 0,
next_message: 0,
loaded_reaction: None,
physical_reads: 0,
finished: false,
}
}
fn load(&mut self, reaction: PhysicalReaction) -> Result<(), BorrowedBatchReceiveFailure> {
let chunk = match reaction.0 {
1 => b"hel".as_slice(),
2 => b"lo\none\ntwo\npar".as_slice(),
3 => b"tial\n".as_slice(),
_ => &[],
};
let Some(end) = self.received_byte_count.checked_add(chunk.len()) else {
return Err(BorrowedBatchReceiveFailure::ReceiveStorageFull);
};
let Some(destination) = self.receive_storage.get_mut(self.received_byte_count..end) else {
return Err(BorrowedBatchReceiveFailure::ReceiveStorageFull);
};
destination.copy_from_slice(chunk);
self.received_byte_count = end;
self.physical_reads = self.physical_reads.saturating_add(1);
while self.next_scan_byte < self.received_byte_count {
let Some(byte) = self.receive_storage.get(self.next_scan_byte) else {
return Err(BorrowedBatchReceiveFailure::ReceiveByteUnavailable {
byte: self.next_scan_byte,
});
};
if *byte == b'\n' {
let Some(range) = self.message_ranges.get_mut(self.message_count) else {
return Err(BorrowedBatchReceiveFailure::MessageCapacityReached);
};
*range = (self.next_message_byte, self.next_scan_byte);
self.message_count = self.message_count.saturating_add(1);
self.next_message_byte = self.next_scan_byte.saturating_add(1);
}
self.next_scan_byte = self.next_scan_byte.saturating_add(1);
}
Ok(())
}
fn decoded_message(&mut self) -> Result<Option<&str>, BorrowedBatchReceiveFailure> {
if self.next_message >= self.message_count {
return Ok(None);
}
let Some(&(start, end)) = self.message_ranges.get(self.next_message) else {
return Err(BorrowedBatchReceiveFailure::MessageCapacityReached);
};
self.next_message = self.next_message.saturating_add(1);
let Some(message) = self.receive_storage.get_mut(start..end) else {
return Err(BorrowedBatchReceiveFailure::MessageRangeUnavailable { start, end });
};
self.decoder
.decode(message)
.map(Some)
.map_err(|source| BorrowedBatchReceiveFailure::InvalidUtf8 { source })
}
fn reset_decoded_batch(&mut self) {
if self.next_message == self.message_count {
self.next_message = 0;
self.message_count = 0;
}
}
}
impl TransportFrom<DecodedMessages> for BorrowedMessageBatchFitting {
type IncomingMessage<'message> = PhysicalReaction;
type Failure = BorrowedBatchReceiveFailure;
fn receive_from<'message>(
&'message mut self,
reaction: Self::IncomingMessage<'message>,
) -> ReceiveFromOutcome<
DecodedMessagesOut<'message>,
Self::IncomingMessage<'message>,
Self::Failure,
> {
if self.next_message < self.message_count {
return match self.decoded_message() {
Ok(Some(message)) => ReceiveFromOutcome::Received {
output: DecodedMessagesOut::Message(message),
},
Ok(None) => ReceiveFromOutcome::Quiescent,
Err(failure) => ReceiveFromOutcome::Failed { failure },
};
}
self.reset_decoded_batch();
if self.loaded_reaction != Some(reaction.0) {
self.loaded_reaction = Some(reaction.0);
if let Err(failure) = self.load(reaction) {
return ReceiveFromOutcome::Failed { failure };
}
if self.next_message < self.message_count {
return match self.decoded_message() {
Ok(Some(message)) => ReceiveFromOutcome::Received {
output: DecodedMessagesOut::Message(message),
},
Ok(None) => ReceiveFromOutcome::Quiescent,
Err(failure) => ReceiveFromOutcome::Failed { failure },
};
}
}
if reaction.0 == 3 && !self.finished {
self.finished = true;
return ReceiveFromOutcome::Received {
output: DecodedMessagesOut::Finished,
};
}
ReceiveFromOutcome::Quiescent
}
}
struct MessageInspector {
message_count: usize,
lengths: [usize; 4],
addresses: [usize; 4],
capacity_reached: bool,
}
impl StateMachine for MessageInspector {}
impl Pipe for MessageInspector {
type In<'message> = InspectMessage<'message>;
type Out<'message> = MessageInspected;
}
struct InspectMessage<'message>(&'message str);
struct MessageInspected;
impl StepInputOf<MessageInspector> for InspectMessage<'_> {
type Outcome = MessageInspected;
fn step(self, inspector: &mut MessageInspector) -> Self::Outcome {
let slot = inspector.message_count;
let Some(length) = inspector.lengths.get_mut(slot) else {
inspector.capacity_reached = true;
return MessageInspected;
};
let Some(address) = inspector.addresses.get_mut(slot) else {
inspector.capacity_reached = true;
return MessageInspected;
};
*length = self.0.len();
*address = self.0.as_ptr() as usize;
inspector.message_count = inspector.message_count.saturating_add(1);
MessageInspected
}
}
struct CompletionSink;
impl Pipe for CompletionSink {
type In<'message> = CompletionSinkIn;
type Out<'message> = NoOutput;
}
enum CompletionSinkIn {
MessageRecorded,
Finish,
}
enum CompletionSinkSubmission {
MessageRecorded,
Finished,
}
struct CompletionSinkFitting {
recorded_message_count: usize,
}
impl TransportTo<CompletionSink> for CompletionSinkFitting {
type Outcome<'transport, 'message> = CompletionSinkSubmission;
fn send<'transport, 'message>(
&'transport mut self,
input: CompletionSinkIn,
) -> Self::Outcome<'transport, 'message> {
match input {
CompletionSinkIn::MessageRecorded => {
self.recorded_message_count = self.recorded_message_count.saturating_add(1);
CompletionSinkSubmission::MessageRecorded
}
CompletionSinkIn::Finish => CompletionSinkSubmission::Finished,
}
}
}
enum FromProgram<'message> {
DecodedMessages(DecodedMessagesOut<'message>),
Inspector(MessageInspected),
}
enum ToProgram<'message> {
Inspector(InspectMessage<'message>),
CompletionSink(CompletionSinkIn),
}
struct ProgramSwitchboard;
impl Switchboard for ProgramSwitchboard {
type From<'message> = FromProgram<'message>;
type To<'message> = ToProgram<'message>;
fn route<'message>(
&mut self,
from: Self::From<'message>,
) -> pipecircuit::Route<Self::To<'message>> {
pipecircuit::Route::one({
match from {
FromProgram::DecodedMessages(DecodedMessagesOut::Message(message)) => {
ToProgram::Inspector(InspectMessage(message))
}
FromProgram::DecodedMessages(DecodedMessagesOut::Finished) => {
ToProgram::CompletionSink(CompletionSinkIn::Finish)
}
FromProgram::Inspector(MessageInspected) => {
ToProgram::CompletionSink(CompletionSinkIn::MessageRecorded)
}
}
})
}
}
struct BatchedProgramCircuit {
reactor: SequenceReactor,
decoded_messages_pipe: InstalledPipe<DecodedMessages, BorrowedMessageBatchFitting>,
inspector_pipe: InstalledPipe<MessageInspector, Inline<MessageInspector>>,
completion_sink_pipe: InstalledPipe<CompletionSink, CompletionSinkFitting>,
}
impl BatchedProgramCircuit {
fn send<'message>(
inspector_pipe: &mut InstalledPipe<MessageInspector, Inline<MessageInspector>>,
completion_sink_pipe: &mut InstalledPipe<CompletionSink, CompletionSinkFitting>,
to: ToProgram<'message>,
) -> Result<CircuitSendOutcome<FromProgram<'message>, ()>, Infallible> {
match to {
ToProgram::Inspector(input) => Ok(CircuitSendOutcome::Immediate {
from: FromProgram::Inspector(inspector_pipe.send(input)),
}),
ToProgram::CompletionSink(input) => match completion_sink_pipe.send(input) {
CompletionSinkSubmission::MessageRecorded => Ok(CircuitSendOutcome::Submitted),
CompletionSinkSubmission::Finished => {
Ok(CircuitSendOutcome::Completed { completion: () })
}
},
}
}
}
impl Circuit<ProgramSwitchboard> for BatchedProgramCircuit {
type Reactor = SequenceReactor;
type Completion = ();
type Failure = BorrowedBatchReceiveFailure;
fn reactor(&mut self) -> &mut Self::Reactor {
&mut self.reactor
}
fn react(
&mut self,
switchboard: &mut ProgramSwitchboard,
reaction: &PhysicalReaction,
) -> Result<CircuitReactionOutcome<Self::Completion>, Self::Failure> {
let Self {
reactor: _,
decoded_messages_pipe,
inspector_pipe,
completion_sink_pipe,
} = self;
match decoded_messages_pipe.receive_from(*reaction) {
ReceiveFromOutcome::Received { output } => {
match flow(
switchboard,
FromProgram::DecodedMessages(output),
&mut pipecircuit::PendingRoutes::<_, 0>::new(),
|to| Self::send(inspector_pipe, completion_sink_pipe, to),
) {
Ok(CircuitFlowOutcome::Submitted) => {
Ok(CircuitReactionOutcome::ContinueReaction)
}
Ok(CircuitFlowOutcome::Completed { completion }) => {
Ok(CircuitReactionOutcome::Completed { completion })
}
Err(pipecircuit::FlowFailure::Dispatch(never)) => match never {},
Err(pipecircuit::FlowFailure::Capacity { .. }) => {
Err(BorrowedBatchReceiveFailure::RoutingCapacity)
}
}
}
ReceiveFromOutcome::Progressed => Ok(CircuitReactionOutcome::ContinueReaction),
ReceiveFromOutcome::Quiescent => Ok(CircuitReactionOutcome::AwaitingReaction),
ReceiveFromOutcome::Backpressured { rejected: _ } => {
Ok(CircuitReactionOutcome::AwaitingReaction)
}
ReceiveFromOutcome::Failed { failure } => Err(failure),
}
}
}
#[test]
fn one_bounded_physical_intake_can_flow_many_borrowed_messages() {
let circuit = BatchedProgramCircuit {
reactor: SequenceReactor { next: 1, waits: 0 },
decoded_messages_pipe: InstalledPipe::new(BorrowedMessageBatchFitting::new()),
inspector_pipe: InstalledPipe::new(Inline::new(MessageInspector {
message_count: 0,
lengths: [0; 4],
addresses: [0; 4],
capacity_reached: false,
})),
completion_sink_pipe: InstalledPipe::new(CompletionSinkFitting {
recorded_message_count: 0,
}),
};
let mut program = Pipecircuit::new(ProgramSwitchboard, circuit);
assert!(program.conduct().is_ok());
let circuit = program.circuit();
let source = circuit.decoded_messages_pipe.transport();
let inspector = circuit.inspector_pipe.transport().destination();
let storage_start = source.receive_storage.as_ptr() as usize;
let storage_end = storage_start + source.received_byte_count;
assert_eq!(circuit.reactor.waits, 3);
assert_eq!(source.physical_reads, 3);
assert_eq!(inspector.message_count, 4);
assert!(!inspector.capacity_reached);
assert_eq!(inspector.lengths, [5, 3, 3, 7]);
assert!(
inspector
.addresses
.iter()
.all(|address| *address >= storage_start && *address < storage_end)
);
assert_eq!(
circuit
.completion_sink_pipe
.transport()
.recorded_message_count,
4
);
}
struct EncodedNumbers;
impl Pipe for EncodedNumbers {
type In<'message> = EncodeNumber;
type Out<'message> = NoOutput;
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct EncodeNumber(u16);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct EncodingDestinationTooSmall;
struct BigEndianU16Encoder;
impl MessageEncoder for BigEndianU16Encoder {
type Message<'message> = u16;
type Error = EncodingDestinationTooSmall;
fn encode<'message>(
&self,
message: &Self::Message<'message>,
destination: &mut [u8],
) -> Result<EncodedByteCount, Self::Error> {
let Some(encoded) = destination.get_mut(..2) else {
return Err(EncodingDestinationTooSmall);
};
encoded.copy_from_slice(&message.to_be_bytes());
Ok(EncodedByteCount::new(2))
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum EncodedBatchSubmission {
Buffered {
message_count: usize,
},
BatchReady {
message_count: usize,
},
Backpressured {
rejected: EncodeNumber,
},
EncodingFailed {
rejected: EncodeNumber,
failure: EncodingDestinationTooSmall,
},
}
struct BoundedEncodedBatchFitting {
encoder: BigEndianU16Encoder,
encoded: [u8; 4],
encoded_byte_count: usize,
message_count: usize,
flush_count: usize,
}
impl BoundedEncodedBatchFitting {
fn pending_bytes(&self) -> Option<&[u8]> {
self.encoded.get(..self.encoded_byte_count)
}
fn complete_flush(&mut self) {
self.encoded_byte_count = 0;
self.message_count = 0;
self.flush_count = self.flush_count.saturating_add(1);
}
}
impl TransportTo<EncodedNumbers> for BoundedEncodedBatchFitting {
type Outcome<'transport, 'message> = EncodedBatchSubmission;
fn send<'transport, 'message>(
&'transport mut self,
input: EncodeNumber,
) -> Self::Outcome<'transport, 'message> {
if self.message_count == 2 {
return EncodedBatchSubmission::Backpressured { rejected: input };
}
let Some(destination) = self.encoded.get_mut(self.encoded_byte_count..) else {
return EncodedBatchSubmission::EncodingFailed {
rejected: input,
failure: EncodingDestinationTooSmall,
};
};
let encoded_byte_count = match self.encoder.encode(&input.0, destination) {
Ok(encoded_byte_count) => encoded_byte_count.get(),
Err(failure) => {
return EncodedBatchSubmission::EncodingFailed {
rejected: input,
failure,
};
}
};
self.encoded_byte_count = self.encoded_byte_count.saturating_add(encoded_byte_count);
self.message_count = self.message_count.saturating_add(1);
if self.message_count == 2 {
EncodedBatchSubmission::BatchReady {
message_count: self.message_count,
}
} else {
EncodedBatchSubmission::Buffered {
message_count: self.message_count,
}
}
}
}
#[test]
fn encoded_output_batches_are_bounded_and_preserve_rejected_work() {
let mut pipe = InstalledPipe::<EncodedNumbers, _>::new(BoundedEncodedBatchFitting {
encoder: BigEndianU16Encoder,
encoded: [0; 4],
encoded_byte_count: 0,
message_count: 0,
flush_count: 0,
});
assert_eq!(
pipe.send(EncodeNumber(7)),
EncodedBatchSubmission::Buffered { message_count: 1 }
);
assert_eq!(
pipe.send(EncodeNumber(9)),
EncodedBatchSubmission::BatchReady { message_count: 2 }
);
assert_eq!(
pipe.send(EncodeNumber(11)),
EncodedBatchSubmission::Backpressured {
rejected: EncodeNumber(11)
}
);
assert_eq!(
pipe.transport().pending_bytes(),
Some([0, 7, 0, 9].as_slice())
);
pipe.transport_mut().complete_flush();
assert_eq!(
pipe.send(EncodeNumber(11)),
EncodedBatchSubmission::Buffered { message_count: 1 }
);
assert_eq!(pipe.transport().pending_bytes(), Some([0, 11].as_slice()));
assert_eq!(pipe.transport().flush_count, 1);
}