fixed tests

examples/calculator.rs

@@ -1,29 +1,18 @@
 use std::{
-    sync::{
-        mpsc::{channel, Receiver, Sender},
-        Mutex,
-    },
+    sync::{mpsc::Receiver, Mutex},
     thread,
     time::Duration,
 };
 
-use seq_ex::{ReplyGuard, SeqEx, SeqNo};
-use std::time::Instant;
+use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync};
 
 #[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),
+enum Packet {
+    Add(f32),
+    Sub(f32),
+    Mul(f32),
+    Div(f32),
+    Mod(f32),
 }
 
 fn drop_packet() -> bool {
@@ -32,43 +21,11 @@ fn drop_packet() -> bool {
     *rng ^= *rng << 13;
     *rng ^= *rng >> 17;
     *rng ^= *rng << 5;
-    *rng & 3 == 0
+    *rng & 1 == 0
 }
 
-struct Transport {
-    channel: Sender<RawPacket>,
-    time: Instant,
-    value: Mutex<f64>,
-}
-
-impl seq_ex::TransportLayer for &Transport {
-    type RecvData = SendPacket;
-
-    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 process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option<Self::SendData>) -> Self::RecvReturn {
-    let mut value = self.value.lock().unwrap();
-    use SendPacket::*;
+fn process(_: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, _: Option<Packet>, value: &mut f32) {
+    use Packet::*;
     match recv_packet {
         Add(n) => *value = *value + n,
         Sub(n) => *value = *value - n,
@@ -78,47 +35,66 @@ fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option<S
     }
 }
 
-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) {}
+fn receive<'a>(
+    recv: &Receiver<PacketType<Packet>>,
+    seq: &SeqExSync<&'a MpscTransport<Packet>>,
+    transport: &'a MpscTransport<Packet>,
+    value: &mut f32,
+) {
+    let packet = recv.try_recv();
+    if !drop_packet() {
+        let do_pump = match packet {
+            Ok(PacketType::Ack { reply_no }) => {
+                seq.receive_ack(reply_no);
+                return;
+            }
+            Ok(PacketType::EmptyReply { reply_no }) => {
+                let result = seq.receive_empty_reply(reply_no);
+                result.is_some()
+            }
+            Ok(PacketType::Data { seq_no, reply_no, payload }) => {
+                if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) {
+                    process(guard, recv_packet, send_packet, value);
+                    true
+                } else {
+                    false
+                }
+            }
+            _ => return,
+        };
+        if do_pump {
+            while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) {
+                process(guard, recv_packet, send_packet, value);
+            }
         }
-        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);
 
+fn main() {
+    let (transport1, recv2) = MpscTransport::new();
+    let (transport2, recv1) = MpscTransport::new();
+    let seq1 = SeqExSync::new(5, 1);
+    let seq2 = SeqExSync::new(5, 1);
     let mut value = 0.0;
-    assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0))));
+    let mut remote_value = value;
+
+    seq1.send(&transport1, Packet::Add(1.0));
     value += 1.0;
-    assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0))));
+    seq1.send(&transport1, Packet::Sub(2.0));
     value -= 2.0;
-    assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0))));
+    seq1.send(&transport1, Packet::Mul(3.0));
     value *= 3.0;
-    assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0))));
+    seq1.send(&transport1, Packet::Div(4.0));
     value /= 4.0;
-    assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0))));
+    seq1.send(&transport1, Packet::Mod(5.0));
     value %= 5.0;
 
     for _ in 0..30 {
-        receive(&recv1, &mut seq1, &transport1);
-        receive(&recv2, &mut seq2, &transport2);
+        receive(&recv1, &seq1, &transport1, &mut value);
+        receive(&recv2, &seq2, &transport2, &mut remote_value);
         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());
+    assert_eq!(value, remote_value);
 }

examples/hello_world.rs

@@ -1,7 +1,6 @@
 use std::sync::mpsc::Receiver;
 
-use seq_ex::sync::{MpscTransport, PacketType};
-use seq_ex::{ReplyGuard, SeqEx};
+use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync};
 
 #[derive(Clone, Debug)]
 enum Packet {
@@ -38,27 +37,27 @@ fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, s
     }
 }
 
-fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &mut SeqEx<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
-    let do_pump = {
-        let result = match recv.recv().unwrap() {
-            PacketType::Ack { reply_no } => {
-                seq.receive_ack(reply_no);
-                return;
+fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
+    let do_pump = match recv.recv().unwrap() {
+        PacketType::Ack { reply_no } => {
+            seq.receive_ack(reply_no);
+            return;
+        }
+        PacketType::EmptyReply { reply_no } => {
+            let result = seq.receive_empty_reply(reply_no);
+            if let Some(Exclamation) = result {
+                // Our Hello World exchange ends right here.
+                print!("\n");
             }
-            PacketType::EmptyReply { reply_no } => {
-                if let Some(Exclamation) = seq.receive_empty_reply(reply_no) {
-                    // Our Hello World exchange ends right here.
-                    print!("\n");
-                }
-                return;
+            result.is_some()
+        }
+        PacketType::Data { seq_no, reply_no, payload } => {
+            if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) {
+                process(guard, recv_packet, send_packet);
+                true
+            } else {
+                false
             }
-            PacketType::Data { seq_no, reply_no, payload } => seq.receive(transport, seq_no, reply_no, payload),
-        };
-        if let Ok((guard, recv_packet, send_packet)) = result {
-            process(guard, recv_packet, send_packet);
-            true
-        } else {
-            false
         }
     };
     if do_pump {
@@ -71,15 +70,15 @@ fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &mut SeqEx<&'a MpscTran
 fn main() {
     let (transport1, recv2) = MpscTransport::new();
     let (transport2, recv1) = MpscTransport::new();
-    let mut seq1 = SeqEx::default();
-    let mut seq2 = SeqEx::default();
+    let seq1 = SeqExSync::default();
+    let seq2 = SeqExSync::default();
 
     // We begin a "Hello World" exchange right here.
-    assert!(seq1.try_send(&transport1, Packet::Hello).is_ok());
+    seq1.send(&transport1, Packet::Hello);
 
-    receive(&recv2, &mut seq2, &transport2);
-    receive(&recv1, &mut seq1, &transport1);
-    receive(&recv2, &mut seq2, &transport2);
-    receive(&recv1, &mut seq1, &transport1);
-    receive(&recv2, &mut seq2, &transport2);
+    receive(&recv2, &seq2, &transport2);
+    receive(&recv1, &seq1, &transport1);
+    receive(&recv2, &seq2, &transport2);
+    receive(&recv1, &seq1, &transport1);
+    receive(&recv2, &seq2, &transport2);
 }

src/seq_queue.rs

@@ -46,8 +46,11 @@ pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 200;
 /// The initial sequence number for a default instance of SeqEx.
 pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1;
 
+pub const DEFAULT_SEND_WINDOW_LEN: usize = 64;
+pub const DEFAULT_RECV_WINDOW_LEN: usize = 32;
+
 const MAX_CONCURRENCY: usize = 24;
-pub struct SeqEx<TL: TransportLayer, const SLEN: usize = 64, const RLEN: usize = 32> {
+pub struct SeqEx<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN, const RLEN: usize = DEFAULT_RECV_WINDOW_LEN> {
     /// The interval at which packets will be resent if they have not yet been acknowledged by the
     /// remote peer.
     /// It can be statically or dynamically set, it is up to the user to decide.
@@ -141,11 +144,11 @@ impl<TL: TransportLayer> SeqEx<TL> {
     }
     /// Sends the given packet to the remote peer and adds it to the send window.
     ///
-    /// If the return value is `false` the queue is full and the packet will not be sent.
+    /// If the return value is `Err` the queue is full and the packet will not be sent.
     /// The caller must either cancel sending, abort the connection, or wait until a call to
-    /// `receive` or `receive_empty_reply` returns `Some` and try again.
+    /// `receive` or `receive_empty_reply` returns `Ok` and try again.
     ///
-    /// If true is returned then the packet was successfully sent.
+    /// If `Ok` is returned then the packet was successfully sent.
     ///
     /// `packet_data` should contain both the packet to be sent as well as any local metadata the
     /// caller wants to store with the packet. This metadata allows the exchange to be stateful.

src/sync.rs

@@ -8,10 +8,12 @@ use std::{
     time::Instant,
 };
 
-use crate::{Error, SeqEx, SeqNo, TransportLayer};
+use crate::{
+    Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RECV_WINDOW_LEN, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_SEND_WINDOW_LEN,
+};
 
-pub struct SeqExSync<TL: TransportLayer> {
-    seq_ex: Mutex<SeqEx<TL>>,
+pub struct SeqExSync<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN, const RLEN: usize = DEFAULT_RECV_WINDOW_LEN> {
+    seq_ex: Mutex<SeqEx<TL, SLEN, RLEN>>,
     /// The mutex above is always held when this value changes, hence it is safe to mutate.
     /// We don't pack this as a component of the mutex to avoid having to reimplement MutexGuard.
     wait_count: UnsafeCell<usize>,
@@ -34,7 +36,15 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
 }
 
 impl<TL: TransportLayer> SeqExSync<TL> {
-    pub fn receive<P: Into<TL::RecvData>, T>(
+    pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
+        Self {
+            seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)),
+            wait_count: UnsafeCell::new(0),
+            send_block: Condvar::default(),
+        }
+    }
+
+    pub fn receive<P: Into<TL::RecvData>>(
         &self,
         app: TL,
         seq_no: SeqNo,
@@ -54,21 +64,11 @@ impl<TL: TransportLayer> SeqExSync<TL> {
     }
     #[inline]
     fn unblock(&self, is_ok: bool) {
-        let has_waiting = unsafe {
-            *self.wait_count.get() > 0
-        };
+        let has_waiting = unsafe { *self.wait_count.get() > 0 };
         if has_waiting && is_ok {
             self.send_block.notify_one();
         }
     }
-
-    pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
-        Self {
-            seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)),
-            wait_count: UnsafeCell::new(0),
-            send_block: Condvar::default(),
-        }
-    }
     pub fn try_send(&self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
         let mut seq = self.lock();
         seq.try_send(app, packet_data)
@@ -77,9 +77,9 @@ impl<TL: TransportLayer> SeqExSync<TL> {
         let mut seq = self.lock();
         while let Err(p) = seq.try_send(app.clone(), packet_data) {
             packet_data = p;
-            unsafe { *self.wait_count.get() += 1}
+            unsafe { *self.wait_count.get() += 1 }
             seq = self.send_block.wait(seq).unwrap();
-            unsafe { *self.wait_count.get() -= 1}
+            unsafe { *self.wait_count.get() -= 1 }
         }
     }
 
@@ -101,6 +101,11 @@ impl<TL: TransportLayer> SeqExSync<TL> {
         self.seq_ex.lock().unwrap()
     }
 }
+impl<TL: TransportLayer> Default for SeqExSync<TL> {
+    fn default() -> Self {
+        Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
+    }
+}
 
 #[derive(Clone)]
 pub enum PacketType<Payload: Clone> {
@@ -139,7 +144,7 @@ impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
         self.time.elapsed().as_millis() as i64
     }
 
-    fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData) {
+    fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Payload) {
         let _ = self.channel.send(PacketType::Data { seq_no, reply_no, payload: payload.clone() });
     }
     fn send_ack(&self, reply_no: SeqNo) {