added examples

examples/calculator.rs

@@ -0,0 +1,131 @@
+use std::{
+    sync::{
+        mpsc::{channel, Receiver, Sender},
+        Mutex,
+    },
+    time::Duration, thread,
+};
+
+use seq_ex::{ReplyGuard, SeqEx, SeqNo};
+use std::time::Instant;
+
+#[derive(Clone)]
+enum RawPacket {
+    Ack(SeqNo),
+    EmptyReply(SeqNo),
+    Send(SeqNo, SendPacket),
+}
+
+#[derive(Clone)]
+enum SendPacket {
+    Add(f64),
+    Sub(f64),
+    Mul(f64),
+    Div(f64),
+    Mod(f64),
+}
+
+fn drop_packet() -> bool {
+    static RNG: Mutex<u32> = Mutex::new(12);
+    let mut rng = RNG.lock().unwrap();
+    *rng ^= *rng << 13;
+    *rng ^= *rng >> 17;
+    *rng ^= *rng << 5;
+    *rng & 1 > 0
+}
+
+struct Transport {
+    channel: Sender<RawPacket>,
+    time: Instant,
+    value: Mutex<f64>,
+}
+
+impl seq_ex::TransportLayer for &Transport {
+    type RecvData = SendPacket;
+    type RecvDataRef<'a> = &'a SendPacket;
+    type RecvReturn = ();
+
+    type SendData = RawPacket;
+
+    fn time(&self) -> i64 {
+        self.time.elapsed().as_millis() as i64
+    }
+
+    fn send(&self, data: &Self::SendData) {
+        if drop_packet() {
+            let _ = self.channel.send(data.clone());
+        }
+    }
+    fn send_ack(&self, reply_no: SeqNo) {
+        if drop_packet() {
+            let _ = self.channel.send(RawPacket::Ack(reply_no));
+        }
+    }
+    fn send_empty_reply(&self, reply_no: SeqNo) {
+        if drop_packet() {
+            let _ = self.channel.send(RawPacket::EmptyReply(reply_no));
+        }
+    }
+
+    fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a> {
+        data
+    }
+    fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option<Self::SendData>) -> Self::RecvReturn {
+        let mut value = self.value.lock().unwrap();
+        use SendPacket::*;
+        match recv_packet {
+            Add(n) => *value = *value + n,
+            Sub(n) => *value = *value - n,
+            Mul(n) => *value = *value * n,
+            Div(n) => *value = *value / n,
+            Mod(n) => *value = *value % n,
+        }
+    }
+}
+
+fn receive<'a>(recv: &Receiver<RawPacket>, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) {
+    match recv.try_recv() {
+        Ok(RawPacket::Ack(reply_no)) => seq.receive_ack(reply_no),
+        Ok(RawPacket::EmptyReply(reply_no)) => {
+            seq.receive_empty_reply(reply_no);
+            while let Ok(()) = seq.pump(transport) {}
+        }
+        Ok(RawPacket::Send(seq_no, packet)) => match seq.receive(transport, seq_no, None, packet) {
+            Ok(()) => {
+                while let Ok(()) = seq.pump(transport) {}
+            }
+            Err(_) => {}
+        }
+        _ => {}
+    }
+}
+fn main() {
+    let (send1, recv1) = channel();
+    let (send2, recv2) = channel();
+    let transport1 = Transport { channel: send2, time: Instant::now(), value: Mutex::new(0.0) };
+    let transport2 = Transport { channel: send1, time: Instant::now(), value: Mutex::new(0.0) };
+    let mut seq1 = SeqEx::new(5, 1);
+    let mut seq2 = SeqEx::new(5, 1);
+
+    let mut value = 0.0;
+    assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0))));
+    value += 1.0;
+    assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0))));
+    value -= 2.0;
+    assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0))));
+    value *= 3.0;
+    assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0))));
+    value /= 4.0;
+    assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0))));
+    value %= 5.0;
+
+    for _ in 0..20 {
+        receive(&recv1, &mut seq1, &transport1);
+        receive(&recv2, &mut seq2, &transport2);
+        thread::sleep(Duration::from_millis(5));
+        seq1.service(&transport1);
+        seq2.service(&transport2);
+    }
+    let remote_value = transport2.value.lock().unwrap();
+    assert_eq!(value, remote_value.clone());
+}

examples/hello_world.rs

@@ -0,0 +1,96 @@
+use std::sync::mpsc::{channel, Receiver, Sender};
+
+use seq_ex::{ReplyGuard, SeqEx, SeqNo};
+use std::time::Instant;
+
+#[derive(Clone)]
+enum RawPacket {
+    Ack(SeqNo),
+    EmptyReply(SeqNo),
+    Send(SeqNo, Packet),
+    Reply(SeqNo, SeqNo, Packet),
+}
+#[derive(Clone)]
+enum Packet {
+    Hello,
+    Space,
+    World,
+    Exclamation,
+}
+
+struct Transport {
+    channel: Sender<RawPacket>,
+    time: Instant,
+}
+
+impl seq_ex::TransportLayer for &Transport {
+    type RecvData = Packet;
+    type RecvDataRef<'a> = &'a Packet;
+    type RecvReturn = ();
+
+    type SendData = RawPacket;
+
+    fn time(&self) -> i64 {
+        self.time.elapsed().as_millis() as i64
+    }
+
+    fn send(&self, data: &Self::SendData) {
+        let _ = self.channel.send(data.clone());
+    }
+    fn send_ack(&self, reply_no: SeqNo) {
+        let _ = self.channel.send(RawPacket::Ack(reply_no));
+    }
+    fn send_empty_reply(&self, reply_no: SeqNo) {
+        let _ = self.channel.send(RawPacket::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::SendData>) -> 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 => {
+                print!(" ");
+                reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::World));
+            }
+            Packet::World => {
+                print!("World");
+                reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::Exclamation));
+            }
+            Packet::Exclamation => {
+                println!("!");
+            }
+        }
+    }
+}
+
+fn receive<'a>(recv: &Receiver<RawPacket>, 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);
+        }
+        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();
+    let transport1 = Transport { channel: send2, time: Instant::now() };
+    let transport2 = Transport { channel: send1, time: Instant::now() };
+    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)));
+
+    receive(&recv2, &mut seq2, &transport2);
+    receive(&recv1, &mut seq1, &transport1);
+    receive(&recv2, &mut seq2, &transport2);
+    receive(&recv1, &mut seq1, &transport1);
+}

src/lib.rs

@@ -55,6 +55,8 @@ pub trait TransportLayer: Sized {
 
     type SendData;
 
+    fn time(&self) -> i64;
+
     fn send(&self, data: &Self::SendData);
     fn send_ack(&self, reply_no: SeqNo);
     fn send_empty_reply(&self, reply_no: SeqNo);
@@ -194,14 +196,14 @@ impl<TL: TransportLayer> SeqEx<TL> {
     /// user would like. However this choice of units must be consistent with the units of the
     /// `retry_interval`. `current_time` does not have to be monotonically increasing.
     #[must_use = "The queue might be full causing the packet to not be sent"]
-    pub fn send(&mut self, app: TL, packet_data: TL::SendData, current_time: i64) -> bool {
+    pub fn send(&mut self, app: TL, packet_data: TL::SendData) -> bool {
         if self.is_full() {
             return false;
         }
         let seq_no = self.next_send_seq_no;
         self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1);
 
-        let next_resent_time = current_time + self.resend_interval;
+        let next_resent_time = app.time() + self.resend_interval;
         let entry = self.send_window[seq_no as usize % self.send_window.len()].insert(SendEntry {
             seq_no,
             reply_no: None,
@@ -213,13 +215,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
         true
     }
 
-    pub fn receive(
-        &mut self,
-        app: TL,
-        seq_no: SeqNo,
-        reply_no: Option<SeqNo>,
-        packet: impl IntoRecvData<TL>,
-    ) -> Result<TL::RecvReturn, Error> {
+    pub fn receive(&mut self, app: TL, seq_no: SeqNo, reply_no: Option<SeqNo>, packet: impl IntoRecvData<TL>) -> Result<TL::RecvReturn, Error> {
         // We only want to accept packets with seq_nos in the range:
         // `self.pre_recv_seq_no < seq_no <= self.pre_recv_seq_no + self.recv_window.len()`.
         // To check that range we compute `seq_no - (self.pre_recv_seq_no + 1)` and check
@@ -328,7 +324,8 @@ impl<TL: TransportLayer> SeqEx<TL> {
         }
     }
 
-    pub fn service(&mut self, app: TL, current_time: i64) -> i64 {
+    pub fn service(&mut self, app: TL) -> i64 {
+        let current_time = app.time();
         let next_interval = current_time + self.resend_interval;
         let mut next_activity = next_interval;
         for item in self.send_window.iter_mut() {
@@ -375,14 +372,14 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
     pub fn reply_no(&self) -> SeqNo {
         self.reply_no
     }
-    pub fn reply(mut self, packet_data: TL::SendData, current_time: i64) {
+    pub fn reply(mut self, packet_data: TL::SendData) {
         if let Some(app) = self.app {
             let seq_queue = &mut self.seq_queue;
             let seq_no = seq_queue.next_send_seq_no;
             seq_queue.next_send_seq_no = seq_queue.next_send_seq_no.wrapping_add(1);
 
             let i = seq_no as usize % seq_queue.send_window.len();
-            let next_resent_time = current_time + seq_queue.resend_interval;
+            let next_resent_time = app.time() + seq_queue.resend_interval;
             let entry = seq_queue.send_window[i].insert(SendEntry {
                 seq_no,
                 reply_no: Some(self.reply_no),
@@ -394,6 +391,10 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
             self.app = None;
         }
     }
+    pub fn reply_with(self, creator: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) {
+        let packet_data = creator(self.seq_no(), self.reply_no());
+        self.reply(packet_data)
+    }
 }
 impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
     fn drop(&mut self) {