diff --git a/examples/hello_world.rs b/examples/hello_world.rs index 1b6fe8e..c3b495f 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -3,32 +3,29 @@ use std::sync::mpsc::{channel, Receiver, Sender}; use seq_ex::{ReplyGuard, SeqEx, SeqNo}; use std::time::Instant; -#[derive(Clone)] -enum RawPacket { +#[derive(Clone, Debug)] +enum Packet { Ack(SeqNo), EmptyReply(SeqNo), - Send(SeqNo, Packet), - Reply(SeqNo, SeqNo, Packet), + Hello(SeqNo), + Reply(SeqNo, SeqNo, ReplyPacket), } -#[derive(Clone)] -enum Packet { - Hello, + +#[derive(Clone, Debug)] +enum ReplyPacket { Space, World, Exclamation, } struct Transport { - channel: Sender, + channel: Sender, time: Instant, } impl seq_ex::TransportLayer for &Transport { type RecvData = Packet; - type RecvDataRef<'a> = &'a Packet; - type RecvReturn = (); - - type SendData = RawPacket; + type SendData = Packet; fn time(&self) -> i64 { self.time.elapsed().as_millis() as i64 @@ -38,47 +35,75 @@ impl seq_ex::TransportLayer for &Transport { let _ = self.channel.send(data.clone()); } fn send_ack(&self, reply_no: SeqNo) { - let _ = self.channel.send(RawPacket::Ack(reply_no)); + let _ = self.channel.send(Packet::Ack(reply_no)); } fn send_empty_reply(&self, reply_no: SeqNo) { - let _ = self.channel.send(RawPacket::EmptyReply(reply_no)); + let _ = self.channel.send(Packet::EmptyReply(reply_no)); } +} - fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a> { - data - } - fn process(&self, reply_cx: ReplyGuard<'_, Self>, recv_packet: &Packet, _: Option) -> Self::RecvReturn { - use RawPacket::Reply; - match recv_packet { - Packet::Hello => { - print!("Hello"); - reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::Space)); - } - Packet::Space => { +fn process(guard: ReplyGuard<'_, &Transport>, recv_packet: Packet, send_packet: Option) { + use Packet::*; + use ReplyPacket::*; + match (recv_packet, send_packet) { + (Hello(_), None) => { + print!("Hello"); + guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Space)); + } + (Reply(_, _, r), Some(p)) => match (r, p) { + (Space, Hello(_)) => { print!(" "); - reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::World)); + guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, World)); } - Packet::World => { + (World, Reply(_, _, Space)) => { print!("World"); - reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::Exclamation)); + guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Exclamation)); } - Packet::Exclamation => { + (Exclamation, Reply(_, _, World)) => { println!("!"); } + (a, b) => { + println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b); + } + }, + (a, None) => { + println!("Unsolicited packet received: {:?}", a); + } + (a, Some(b)) => { + println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b); } } } -fn receive<'a>(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) { - match recv.recv().unwrap() { - RawPacket::Ack(reply_no) => seq.receive_ack(reply_no), - RawPacket::EmptyReply(reply_no) => { - seq.receive_empty_reply(reply_no); +fn receive<'a>(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) { + use Packet::*; + let do_pump = { + let result = match recv.recv().unwrap() { + Ack(reply_no) => { + seq.receive_ack(reply_no); + return; + } + EmptyReply(reply_no) => { + seq.receive_empty_reply(reply_no); + return; + } + Hello(seq_no) => seq.receive(transport, seq_no, None, Hello(seq_no)), + Reply(seq_no, reply_no, p) => seq.receive(transport, seq_no, Some(reply_no), Reply(seq_no, reply_no, p)), + }; + if let Ok((guard, recv_packet, send_packet)) = result { + process(guard, recv_packet, send_packet); + true + } else { + false + } + }; + if do_pump { + while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) { + process(guard, recv_packet, send_packet); } - RawPacket::Send(seq_no, packet) => seq.receive(transport, seq_no, None, packet).unwrap(), - RawPacket::Reply(seq_no, reply_no, packet) => seq.receive(transport, seq_no, Some(reply_no), packet).unwrap(), } } + fn main() { let (send1, recv1) = channel(); let (send2, recv2) = channel(); @@ -87,7 +112,7 @@ fn main() { let mut seq1 = SeqEx::new(100, 1); let mut seq2 = SeqEx::new(100, 1); - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), Packet::Hello))); + assert!(seq1.send_with(&transport1, |seq_no| Packet::Hello(seq_no))); receive(&recv2, &mut seq2, &transport2); receive(&recv1, &mut seq1, &transport1); diff --git a/src/multi_thread.rs b/src/multi_thread.rs index 6b8b370..ed1bd77 100644 --- a/src/multi_thread.rs +++ b/src/multi_thread.rs @@ -6,7 +6,7 @@ pub struct SeqExLock(pub Mutex>); pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExLock, TL, SeqNo); impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { - pub fn reply(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { let mut seq = self.0 .0.lock().unwrap(); let p = packet_data(seq.seq_no(), self.2); seq.reply_raw(self.1.clone(), self.2, p); diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 275e5ce..2c97f6e 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -169,6 +169,11 @@ impl SeqEx { app.send(&entry.data); true } + #[must_use = "The queue might be full causing the packet to not be sent"] + pub fn send_with(&mut self, app: TL, create_data: impl FnOnce(SeqNo) -> TL::SendData) -> bool { + let p = create_data(self.seq_no()); + self.send(app, p) + } pub fn receive_raw>( &mut self, diff --git a/src/single_thread.rs b/src/single_thread.rs index 40a09cf..9f5e4ce 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,6 +12,10 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { self.0.reply_raw(self.1.clone(), self.2, packet_data); core::mem::forget(self); } + pub fn reply_with(self, create_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { + let p = create_data(self.seq_no(), self.reply_no()); + self.reply(p) + } } impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { fn drop(&mut self) { @@ -20,7 +24,7 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { } impl SeqEx { - pub fn receive, T>( + pub fn receive>( &mut self, app: TL, seq_no: SeqNo,