added file download example

Cargo.lock

@@ -19,12 +19,36 @@ dependencies = [
  "wasi",
 ]
 
+[[package]]
+name = "itoa"
+version = "1.0.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af150ab688ff2122fcef229be89cb50dd66af9e01a4ff320cc137eecc9bacc38"
+
 [[package]]
 name = "libc"
 version = "0.2.147"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3"
 
+[[package]]
+name = "proc-macro2"
+version = "1.0.66"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18fb31db3f9bddb2ea821cde30a9f70117e3f119938b5ee630b7403aa6e2ead9"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "50f3b39ccfb720540debaa0164757101c08ecb8d326b15358ce76a62c7e85965"
+dependencies = [
+ "proc-macro2",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.6.4"
@@ -34,13 +58,69 @@ dependencies = [
  "getrandom",
 ]
 
+[[package]]
+name = "ryu"
+version = "1.0.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ad4cc8da4ef723ed60bced201181d83791ad433213d8c24efffda1eec85d741"
+
 [[package]]
 name = "seq_ex"
 version = "0.1.0"
 dependencies = [
  "rand_core",
+ "serde",
+ "serde_json",
 ]
 
+[[package]]
+name = "serde"
+version = "1.0.183"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32ac8da02677876d532745a130fc9d8e6edfa81a269b107c5b00829b91d8eb3c"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.183"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aafe972d60b0b9bee71a91b92fee2d4fb3c9d7e8f6b179aa99f27203d99a4816"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.105"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "693151e1ac27563d6dbcec9dee9fbd5da8539b20fa14ad3752b2e6d363ace360"
+dependencies = [
+ "itoa",
+ "ryu",
+ "serde",
+]
+
+[[package]]
+name = "syn"
+version = "2.0.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "04361975b3f5e348b2189d8dc55bc942f278b2d482a6a0365de5bdd62d351567"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "301abaae475aa91687eb82514b328ab47a211a533026cb25fc3e519b86adfc3c"
+
 [[package]]
 name = "wasi"
 version = "0.11.0+wasi-snapshot-preview1"

Cargo.toml

@@ -10,8 +10,12 @@ path = "src/lib.rs"
 doc = true
 
 [features]
-default = ["std"]
+default = ["std", "serde"]
 std = []
 
+[dependencies]
+serde = { version = "1.0.183", default-features = false, features = ["derive"], optional = true }
+
 [dev-dependencies]
 rand_core = { version = "0.6.4", features = ["getrandom"]}
+serde_json = { version = "1.0.104" }

examples/calculator.rs

@@ -1,6 +1,6 @@
 use std::{sync::mpsc::Receiver, thread, time::Duration};
 
-use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
+use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, MpscGuard, MpscSeqEx};
 
 #[derive(Clone)]
 enum Packet {
@@ -16,7 +16,7 @@ fn drop_packet() -> bool {
     rand_core::OsRng.next_u32() & 1 > 0
 }
 
-fn process(_: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, _: Option<Packet>, value: &mut f32) {
+fn process(_: MpscGuard<'_, Packet>, recv_packet: Packet, _: Option<Packet>, value: &mut f32) {
     use Packet::*;
     match recv_packet {
         Add(n) => *value = *value + n,
@@ -29,17 +29,17 @@ fn process(_: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, _: Op
 
 fn receive<'a>(
     recv: &Receiver<PacketType<Packet>>,
-    seq: &SeqExSync<&'a MpscTransport<Packet>>,
+    seq: &MpscSeqEx<Packet>,
     transport: &'a MpscTransport<Packet>,
     value: &mut f32,
 ) {
     while let Ok(packet) = recv.try_recv() {
         if !drop_packet() {
             match packet {
-                PacketType::Ack { reply_no } => {
+                PacketType::Ack ( reply_no ) => {
                     let _ = seq.receive_ack(reply_no);
                 }
-                PacketType::Payload { seq_no, reply_no, payload } => {
+                PacketType::Payload ( seq_no, reply_no, payload ) => {
                     for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) {
                         process(guard, packet, send_data, value);
                     }
@@ -52,8 +52,8 @@ fn receive<'a>(
 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 seq1 = MpscSeqEx::new(5, 1);
+    let seq2 = MpscSeqEx::new(5, 1);
     let mut value = 0.0;
     let mut remote_value = value;
 
@@ -68,7 +68,7 @@ fn main() {
     seq1.send(&transport1, Packet::Mod(5.0));
     value %= 5.0;
 
-    for _ in 0..30 {
+    for _ in 0..16 {
         receive(&recv1, &seq1, &transport1, &mut value);
         receive(&recv2, &seq2, &transport2, &mut remote_value);
         thread::sleep(Duration::from_millis(5));

examples/file_download.rs

@@ -0,0 +1,157 @@
+use std::{sync::{mpsc::{Receiver, Sender, channel}, Arc, RwLock}, thread, time::{Duration, Instant}, collections::HashMap, ops::Deref};
+
+use seq_ex::{sync::{PacketType, RecvSuccess, ReplyGuard, SeqExSync}, TransportLayer, SeqNo};
+use rand_core::{RngCore, OsRng};
+use serde::{Serialize, Deserialize};
+
+const FILE_CHUNK_SIZE: usize = 1000;
+#[derive(Clone, Debug, Serialize, Deserialize)]
+enum Packet {
+    RequestFile {
+        filename: String,
+    },
+    ConfirmRequestFile {
+        filesize: u64,
+    },
+    FileDownload {
+        filename: String,
+        file_chunk: Vec<u8>,
+    }
+}
+
+#[derive(Clone)]
+struct Transport {
+    sender: Sender<Vec<u8>>,
+    time: Instant,
+}
+struct Peer {
+    filesystem: Arc<RwLock<HashMap<String, Vec<u8>>>>,
+    transport: Transport,
+    seqex: Arc<SeqExSync<Packet, Packet>>,
+    receiver: Receiver<Vec<u8>>,
+}
+
+impl TransportLayer for &Transport {
+    type SendData = Packet;
+
+    fn time(&mut self) -> i64 {
+        self.time.elapsed().as_millis() as i64
+    }
+
+    fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Packet) {
+        let p = PacketType::Payload(seq_no, reply_no, payload.clone());
+        if let Ok(p) = serde_json::to_vec(&p) {
+            let _ = self.sender.send(p);
+        }
+    }
+
+    fn send_ack(&mut self, reply_no: SeqNo) {
+        let p = PacketType::<Packet>::Ack(reply_no);
+        if let Ok(p) = serde_json::to_vec(&p) {
+            let _ = self.sender.send(p);
+        }
+    }
+}
+
+
+fn drop_packet() -> bool {
+    OsRng.next_u32() >= (u32::MAX / 4 * 3)
+}
+
+fn process<'a>(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_packet: Packet, sent_packet: Option<Packet>) {
+    match (recv_packet, sent_packet) {
+        (Packet::RequestFile { filename }, None) => {
+            let filesystem = peer.filesystem.clone();
+            let transport = peer.transport.clone();
+            let seqex = peer.seqex.clone();
+            if let Some(file) = filesystem.read().unwrap().get(&filename) {
+                guard.reply(Packet::ConfirmRequestFile { filesize: file.len() as u64 });
+            }
+            thread::spawn(move || {
+                let filesystem = filesystem.read().unwrap();
+                // NOTE: in a real application you need to explicitly handle the situation where the
+                // file is missing.
+                if let Some(file) = filesystem.get(&filename) {
+                    let mut i = 0;
+                    while i < file.len() {
+                        let j = file.len().min(i + FILE_CHUNK_SIZE);
+                        seqex.send(&transport, Packet::FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() });
+                        i = j;
+                    }
+                }
+            });
+        }
+        (Packet::ConfirmRequestFile { filesize }, Some(Packet::RequestFile { filename })) => {
+            let mut filesystem = peer.filesystem.write().unwrap();
+            let file = Vec::with_capacity(filesize as usize);
+            filesystem.insert(filename, file);
+        }
+        (Packet::FileDownload { filename, file_chunk }, None) => {
+            let mut filesystem = peer.filesystem.write().unwrap();
+            if let Some(file) = filesystem.get_mut(&filename) {
+                if file.len() + file_chunk.len() <= file.capacity() {
+                    file.extend(&file_chunk);
+                }
+            }
+        }
+        _ => {
+            assert!(false);
+        }
+    }
+}
+
+fn receive<'a>(
+    peer: &Peer,
+) {
+    while let Ok(packet) = peer.receiver.try_recv() {
+        if drop_packet() {
+            continue;
+        }
+        let parsed_packet = serde_json::from_slice::<PacketType<Packet>>(&packet);
+        match parsed_packet {
+            Ok(PacketType::Ack ( reply_no )) => {
+                let _ = peer.seqex.receive_ack(reply_no);
+            }
+            Ok(PacketType::Payload ( seq_no, reply_no, payload )) => {
+                for RecvSuccess { guard, packet, send_data } in peer.seqex.receive_all(&peer.transport, seq_no, reply_no, payload) {
+                    process(peer, guard, packet, send_data);
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
+
+fn main() {
+    let mut filesystem2 = HashMap::new();
+    let mut file = Vec::from([0u8; 1 << 16]);
+    OsRng.fill_bytes(&mut file);
+    filesystem2.insert("File1".to_string(), file);
+    let mut file = Vec::from([0u8; 1 << 18]);
+    OsRng.fill_bytes(&mut file);
+    filesystem2.insert("File2".to_string(), file);
+    let mut file = Vec::from([0u8; 1 << 20]);
+    OsRng.fill_bytes(&mut file);
+    filesystem2.insert("File3".to_string(), file);
+
+    let (send1, recv2) = channel();
+    let (send2, recv1) = channel();
+
+    let peer1 = Peer { filesystem: Arc::new(RwLock::new(HashMap::new())), seqex: Arc::new(SeqExSync::new(5, 1)), transport: Transport{time: Instant::now(), sender: send1}, receiver: recv1 };
+    let peer2 = Peer { filesystem: Arc::new(RwLock::new(filesystem2)), seqex: Arc::new(SeqExSync::new(5, 1)), transport: Transport{time: Instant::now(), sender: send2}, receiver: recv2 };
+
+    peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File1".to_string()});
+    peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File3".to_string()});
+    peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File2".to_string()});
+
+    for _ in 0..300 {
+        receive(&peer1);
+        receive(&peer2);
+        thread::sleep(Duration::from_millis(1));
+        peer1.seqex.service(&peer1.transport);
+        peer2.seqex.service(&peer2.transport);
+    }
+
+    assert_eq!(peer1.filesystem.read().unwrap().deref(), peer2.filesystem.read().unwrap().deref());
+}

examples/hello_world.rs

@@ -1,6 +1,6 @@
 use std::sync::mpsc::Receiver;
 
-use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
+use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, MpscGuard, MpscSeqEx};
 
 #[derive(Clone, Debug)]
 enum Packet {
@@ -11,7 +11,7 @@ enum Packet {
 }
 use Packet::*;
 
-fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, send_packet: Option<Packet>) {
+fn process(guard: MpscGuard<'_, Packet>, recv_packet: Packet, send_packet: Option<Packet>) {
     match (recv_packet, send_packet) {
         (Hello, None) => {
             print!("Hello");
@@ -37,16 +37,16 @@ fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, s
     }
 }
 
-fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
+fn receive(recv: &Receiver<PacketType<Packet>>, seq: &MpscSeqEx<Packet>, transport: &MpscTransport<Packet>) {
     match recv.recv().unwrap() {
-        PacketType::Ack { reply_no } => {
+        PacketType::Ack ( reply_no ) => {
             let result = seq.receive_ack(reply_no);
             if let Ok(Exclamation) = result {
                 // Our Hello World exchange ends right here.
                 print!("\n");
             }
         }
-        PacketType::Payload { seq_no, reply_no, payload } => {
+        PacketType::Payload ( seq_no, reply_no, payload ) => {
             for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) {
                 process(guard, packet, send_data)
             }
@@ -57,8 +57,8 @@ fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTran
 fn main() {
     let (transport1, recv2) = MpscTransport::new();
     let (transport2, recv1) = MpscTransport::new();
-    let seq1 = SeqExSync::default();
-    let seq2 = SeqExSync::default();
+    let seq1 = MpscSeqEx::default();
+    let seq2 = MpscSeqEx::default();
 
     // We begin a "Hello World" exchange right here.
     seq1.send(&transport1, Packet::Hello);

src/seq_queue.rs

@@ -48,7 +48,7 @@ pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1;
 
 pub const DEFAULT_WINDOW_CAP: usize = 64;
 
-pub struct SeqEx<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
+pub struct SeqEx<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     /// 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.
@@ -57,8 +57,8 @@ pub struct SeqEx<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
     next_send_seq_no: SeqNo,
     pre_recv_seq_no: SeqNo,
     /// This could be made more efficient by changing to SoA format.
-    send_window: [Option<SendEntry<TL>>; CAP],
-    recv_window: [Option<RecvEntry<TL>>; CAP],
+    send_window: [Option<SendEntry<SendData>>; CAP],
+    recv_window: [Option<RecvEntry<RecvData>>; CAP],
     /// The size of this array determines the maximum number of received packets that the application
     /// may attempt to process concurrently before new received packets start being dropped.
     concurrent_replies: [SeqNo; CAP],
@@ -69,17 +69,17 @@ pub struct SeqEx<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
     concurrent_replies_total: usize,
 }
 
-struct RecvEntry<TL: TransportLayer> {
+struct RecvEntry<RecvData> {
     seq_no: SeqNo,
     reply_no: Option<SeqNo>,
-    data: TL::RecvData,
+    data: RecvData,
 }
 
-struct SendEntry<TL: TransportLayer> {
+struct SendEntry<SendData> {
     seq_no: SeqNo,
     reply_no: Option<SeqNo>,
     next_resend_time: i64,
-    data: TL::SendData,
+    data: SendData,
 }
 
 /// The error type for when a packet has been received, but for whatever reason could not be
@@ -97,7 +97,7 @@ pub enum Error {
 /// An iterator over all packets in the send window. It will iterate over all packets currently
 /// being sent to the remote peer.
 /// These packets are awaiting a reply from the remote peer.
-pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option<SendEntry<TL>>>);
+pub struct Iter<'a, SendData>(core::slice::Iter<'a, Option<SendEntry<SendData>>>);
 /// A mutable iterator over all packets in the send window.
 ///
 /// The user is able to mutate the contents of the packet being sent to the remote peer, as well as
@@ -107,9 +107,9 @@ pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option<SendEntry<T
 /// version of the packet will have been received by the remote peer. The local peer cannot be sure
 /// if the remote peer will see the modified packet. For this reason it is not recommended to modify
 /// the packet.
-pub struct IterMut<'a, TL: TransportLayer>(core::slice::IterMut<'a, Option<SendEntry<TL>>>);
+pub struct IterMut<'a, SendData>(core::slice::IterMut<'a, Option<SendEntry<SendData>>>);
 
-impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
+impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// Creates a new instance of `SeqEx` for a new remote peer.
     /// An instance of `SeqEx` expects to communicate with only exactly one other remote instance
     /// of `SeqEx`.
@@ -131,10 +131,10 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
             concurrent_replies_total: 0,
         }
     }
-    fn send_window_slot_mut(&mut self, seq_no: SeqNo) -> &mut Option<SendEntry<TL>> {
+    fn send_window_slot_mut(&mut self, seq_no: SeqNo) -> &mut Option<SendEntry<SendData>> {
         &mut self.send_window[seq_no as usize % self.send_window.len()]
     }
-    fn send_window_slot(&self, seq_no: SeqNo) -> &Option<SendEntry<TL>> {
+    fn send_window_slot(&self, seq_no: SeqNo) -> &Option<SendEntry<SendData>> {
         &self.send_window[seq_no as usize % self.send_window.len()]
     }
 
@@ -171,7 +171,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
     /// 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 try_send(&mut self, mut app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
+    pub fn try_send(&mut self, mut app: impl TransportLayer<SendData = SendData>, packet_data: SendData) -> Result<(), SendData> {
         if self.is_full() {
             return Err(packet_data);
         }
@@ -192,13 +192,13 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
         Ok(())
     }
 
-    pub fn receive_raw<P: Into<TL::RecvData>>(
+    pub fn receive_raw<P: Into<RecvData>>(
         &mut self,
-        mut app: TL,
+        mut app: impl TransportLayer<SendData = SendData>,
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
-    ) -> Result<(SeqNo, P, Option<TL::SendData>), Error> {
+    ) -> Result<(SeqNo, P, Option<SendData>), Error> {
         // We only want to accept packets with sequence numbers 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
@@ -270,7 +270,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
             }
         }
     }
-    pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result<TL::SendData, Error> {
+    pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result<SendData, Error> {
         let slot = self.send_window_slot_mut(reply_no);
         if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) {
             let entry = slot.take().unwrap();
@@ -293,7 +293,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
         false
     }
 
-    fn take_send(&mut self, reply_no: SeqNo) -> Option<TL::SendData> {
+    fn take_send(&mut self, reply_no: SeqNo) -> Option<SendData> {
         let slot = self.send_window_slot_mut(reply_no);
         if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) {
             slot.take().map(|e| e.data)
@@ -301,7 +301,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
             None
         }
     }
-    pub fn pump_raw(&mut self) -> Result<(SeqNo, TL::RecvData, Option<TL::SendData>), Error> {
+    pub fn pump_raw(&mut self) -> Result<(SeqNo, RecvData, Option<SendData>), Error> {
         let next_seq_no = self.pre_recv_seq_no.wrapping_add(1);
         let i = next_seq_no as usize % self.recv_window.len();
 
@@ -329,7 +329,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
     /// The identifier will tell the remote peer which packets contain fragments of the file,
     /// and since each fragment will be received in order it will be trivial for them to reconstruct
     /// the original file.
-    pub fn reply_raw(&mut self, mut app: TL, reply_no: SeqNo, packet_data: TL::SendData) {
+    pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData = SendData>, reply_no: SeqNo, packet_data: SendData) {
         if self.remove_reservation(reply_no) {
             let seq_no = self.next_send_seq_no;
             self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1);
@@ -352,7 +352,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
             app.send(entry.seq_no, entry.reply_no, &entry.data);
         }
     }
-    pub fn ack_raw(&mut self, mut app: TL, reply_no: SeqNo) {
+    pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData = SendData>, reply_no: SeqNo) {
         if self.remove_reservation(reply_no) {
             // Acks are only sent once. There is code in `receive_raw` to handle resending
             // an ack in the event that the first one here was dropped by the network.
@@ -371,7 +371,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
         false
     }
 
-    pub fn service(&mut self, mut app: TL) -> i64 {
+    pub fn service(&mut self, mut app: impl TransportLayer<SendData = SendData>) -> i64 {
         let current_time = app.time();
         let next_interval = current_time + self.resend_interval;
         let mut next_activity = i64::MAX;
@@ -388,29 +388,29 @@ impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
         self.resend_interval.min(next_activity - current_time)
     }
 
-    pub fn iter(&self) -> Iter<'_, TL> {
+    pub fn iter(&self) -> Iter<'_, SendData> {
         Iter(self.send_window.iter())
     }
-    pub fn iter_mut(&mut self) -> IterMut<'_, TL> {
+    pub fn iter_mut(&mut self) -> IterMut<'_, SendData> {
         IterMut(self.send_window.iter_mut())
     }
 }
-impl<TL: TransportLayer> Default for SeqEx<TL> {
+impl<SendData, RecvData, const CAP: usize> Default for SeqEx<SendData, RecvData, CAP> {
     fn default() -> Self {
         Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
     }
 }
-impl<'a, TL: TransportLayer> IntoIterator for &'a SeqEx<TL> {
-    type Item = &'a TL::SendData;
-    type IntoIter = Iter<'a, TL>;
+impl<'a, SendData, RecvData, const CAP: usize> IntoIterator for &'a SeqEx<SendData, RecvData, CAP> {
+    type Item = &'a SendData;
+    type IntoIter = Iter<'a, SendData>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.iter()
     }
 }
-impl<'a, TL: TransportLayer> IntoIterator for &'a mut SeqEx<TL> {
-    type Item = &'a mut TL::SendData;
-    type IntoIter = IterMut<'a, TL>;
+impl<'a, SendData, RecvData, const CAP: usize> IntoIterator for &'a mut SeqEx<SendData, RecvData, CAP> {
+    type Item = &'a mut SendData;
+    type IntoIter = IterMut<'a, SendData>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.iter_mut()
@@ -419,8 +419,8 @@ impl<'a, TL: TransportLayer> IntoIterator for &'a mut SeqEx<TL> {
 
 macro_rules! iterator {
     ($iter:ident, {$( $mut:tt )?}) => {
-        impl<'a, TL: TransportLayer> Iterator for $iter<'a, TL> {
-            type Item = &'a $($mut)? TL::SendData;
+        impl<'a, SendData> Iterator for $iter<'a, SendData> {
+            type Item = &'a $($mut)? SendData;
             fn next(&mut self) -> Option<Self::Item> {
                 while let Some(entry) = self.0.next() {
                     if let Some(entry) = entry {
@@ -434,7 +434,7 @@ macro_rules! iterator {
                 (0, Some(self.0.len()))
             }
         }
-        impl<'a, TL: TransportLayer> DoubleEndedIterator for $iter<'a, TL> {
+        impl<'a, SendData> DoubleEndedIterator for $iter<'a, SendData> {
             fn next_back(&mut self) -> Option<Self::Item> {
                 while let Some(entry) = self.0.next_back() {
                     if let Some(entry) = entry {

src/single_thread.rs

@@ -1,41 +1,41 @@
 use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
 
-pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a mut SeqEx<TL, CAP>, TL, SeqNo);
-impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
+pub struct ReplyGuard<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a mut SeqEx<SendData, RecvData, CAP>, TL, SeqNo);
+impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
     /// If you need to reply more than once, say to fragment a large file, then include in your
     /// first reply some identifier, and then `send` all fragments with the same included identifier.
     /// The identifier will tell the remote peer which packets contain fragments of the file,
     /// and since each fragment will be received in order it will be trivial for them to reconstruct
     /// the original file.
-    pub fn reply(self, packet_data: TL::SendData) {
+    pub fn reply(self, packet_data: SendData) {
         self.0.reply_raw(self.1.clone(), self.2, packet_data);
         core::mem::forget(self);
     }
 }
-impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> {
+impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
     fn drop(&mut self) {
         self.0.ack_raw(self.1.clone(), self.2)
     }
 }
 
-pub struct RecvSuccess<'a, TL: TransportLayer, P, const CAP: usize = DEFAULT_WINDOW_CAP> {
-    pub guard: ReplyGuard<'a, TL, CAP>,
+pub struct RecvSuccess<'a, TL: TransportLayer<SendData = SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+    pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
     pub packet: P,
-    pub send_data: Option<TL::SendData>,
+    pub send_data: Option<SendData>,
 }
 
-impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
-    pub fn receive<P: Into<TL::RecvData>>(
+impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
+    pub fn receive<TL: TransportLayer<SendData = SendData>, P: Into<RecvData>>(
         &mut self,
         app: TL,
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
-    ) -> Result<RecvSuccess<'_, TL, P, CAP>, Error> {
+    ) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, CAP>, Error> {
         self.receive_raw(app.clone(), seq_no, reply_no, packet)
             .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
-    pub fn pump(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData, CAP>, Error> {
+    pub fn pump<TL: TransportLayer<SendData = SendData>>(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
         self.pump_raw()
             .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }

src/sync.rs

@@ -10,75 +10,75 @@ use std::{
 
 use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
 
-pub struct SeqExSync<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
-    seq_ex: Mutex<SeqEx<TL, CAP>>,
+pub struct SeqExSync<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+    seq_ex: Mutex<SeqEx<SendData, RecvData, CAP>>,
     /// 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>,
     send_block: Condvar,
 }
 
-pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync<TL, CAP>, TL, SeqNo);
-impl<'a, TL: TransportLayer, const CAP: usize> ReplyGuard<'a, TL, CAP> {
+pub struct ReplyGuard<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync<SendData, RecvData, CAP>, TL, SeqNo);
+impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
     /// If you need to reply more than once, say to fragment a large file, then include in your
     /// first reply some identifier, and then `send` all fragments with the same included identifier.
     /// The identifier will tell the remote peer which packets contain fragments of the file,
     /// and since each fragment will be received in order it will be trivial for them to reconstruct
     /// the original file.
-    pub fn reply(self, packet_data: TL::SendData) {
+    pub fn reply(self, packet_data: SendData) {
         let mut seq = self.0.seq_ex.lock().unwrap();
         seq.reply_raw(self.1.clone(), self.2, packet_data);
         core::mem::forget(self);
     }
 }
-impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> {
+impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
     fn drop(&mut self) {
         let mut seq = self.0.seq_ex.lock().unwrap();
         seq.ack_raw(self.1.clone(), self.2);
     }
 }
 
-pub struct RecvSuccess<'a, TL: TransportLayer, P: Into<TL::RecvData>, const CAP: usize = DEFAULT_WINDOW_CAP> {
-    pub guard: ReplyGuard<'a, TL, CAP>,
+pub struct RecvSuccess<'a, TL: TransportLayer<SendData = SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+    pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
     pub packet: P,
-    pub send_data: Option<TL::SendData>,
+    pub send_data: Option<SendData>,
 }
 
-pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
-    origin: Option<&'a SeqExSync<TL, CAP>>,
+pub struct ReplyIter<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+    origin: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
     app: TL,
-    first: Option<RecvSuccess<'a, TL, TL::RecvData, CAP>>,
+    first: Option<RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>>,
 }
 
-impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
+impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
     pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
         Self {
-            seq_ex: Mutex::new(SeqEx::<TL, CAP>::new(retry_interval, initial_seq_no)),
+            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>>(
+    pub fn receive<TL: TransportLayer<SendData = SendData>, P: Into<RecvData>>(
         &self,
         app: TL,
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
-    ) -> Result<RecvSuccess<'_, TL, P, CAP>, Error> {
+    ) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, CAP>, Error> {
         let mut seq = self.lock();
         let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet);
         // TODO: double check blocking.
         self.unblock(ret.is_ok());
         ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
-    pub fn pump(&self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData, CAP>, Error> {
+    pub fn pump<TL: TransportLayer<SendData = SendData>>(&self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
         let mut seq = self.lock();
         let ret = seq.pump_raw();
         self.unblock(ret.is_ok());
         ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
-    pub fn receive_all(&self, app: TL, seq_no: SeqNo, reply_no: Option<SeqNo>, packet: TL::RecvData) -> ReplyIter<'_, TL, CAP> {
+    pub fn receive_all<TL: TransportLayer<SendData = SendData>>(&self, app: TL, seq_no: SeqNo, reply_no: Option<SeqNo>, packet: RecvData) -> ReplyIter<'_, TL, SendData, RecvData, CAP> {
         if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) {
             ReplyIter { origin: Some(self), app, first: Some(g) }
         } else {
@@ -92,11 +92,11 @@ impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
             self.send_block.notify_one();
         }
     }
-    pub fn try_send(&self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
+    pub fn try_send<TL: TransportLayer<SendData = SendData>>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> {
         let mut seq = self.lock();
         seq.try_send(app, packet_data)
     }
-    pub fn send(&self, app: TL, mut packet_data: TL::SendData) {
+    pub fn send<TL: TransportLayer<SendData = SendData>>(&self, app: TL, mut packet_data: SendData) {
         let mut seq = self.lock();
         while let Err(p) = seq.try_send(app.clone(), packet_data) {
             packet_data = p;
@@ -106,27 +106,29 @@ impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
         }
     }
 
-    pub fn receive_ack(&self, reply_no: SeqNo) -> Result<TL::SendData, Error> {
+    pub fn receive_ack(&self, reply_no: SeqNo) -> Result<SendData, Error> {
         let ret = self.lock().receive_ack(reply_no);
         self.unblock(ret.is_ok());
         ret
     }
-    pub fn service(&self, app: TL) -> i64 {
+    pub fn service<TL: TransportLayer<SendData = SendData>>(&self, app: TL) -> i64 {
         self.lock().service(app)
     }
 
-    pub fn lock(&self) -> MutexGuard<SeqEx<TL, CAP>> {
+    pub fn lock(&self) -> MutexGuard<SeqEx<SendData, RecvData, CAP>> {
         self.seq_ex.lock().unwrap()
     }
 }
-impl<TL: TransportLayer, const CAP: usize> Default for SeqExSync<TL, CAP> {
+impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvData, CAP> {
     fn default() -> Self {
         Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
     }
 }
+unsafe impl<SendData, RecvData, const CAP: usize> Send for SeqExSync<SendData, RecvData, CAP> {}
+unsafe impl<SendData, RecvData, const CAP: usize> Sync for SeqExSync<SendData, RecvData, CAP> {}
 
-impl<'a, TL: TransportLayer, const CAP: usize> Iterator for ReplyIter<'a, TL, CAP> {
-    type Item = RecvSuccess<'a, TL, TL::RecvData, CAP>;
+impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> {
+    type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>;
     fn next(&mut self) -> Option<Self::Item> {
         if let Some(g) = self.first.take() {
             Some(g)
@@ -138,16 +140,11 @@ impl<'a, TL: TransportLayer, const CAP: usize> Iterator for ReplyIter<'a, TL, CA
     }
 }
 
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 #[derive(Clone)]
 pub enum PacketType<Payload: Clone> {
-    Payload {
-        seq_no: SeqNo,
-        reply_no: Option<SeqNo>,
-        payload: Payload,
-    },
-    Ack {
-        reply_no: SeqNo,
-    },
+    Payload(SeqNo, Option<SeqNo>, Payload),
+    Ack(SeqNo),
 }
 
 #[derive(Clone)]
@@ -155,6 +152,9 @@ pub struct MpscTransport<Payload: Clone> {
     pub channel: Sender<PacketType<Payload>>,
     pub time: Instant,
 }
+pub type MpscGuard<'a, Packet> = ReplyGuard<'a, &'a MpscTransport<Packet>, Packet, Packet>;
+pub type MpscSeqEx<Packet> = SeqExSync<Packet, Packet>;
+
 impl<Payload: Clone> MpscTransport<Payload> {
     pub fn new() -> (Self, Receiver<PacketType<Payload>>) {
         let (send, recv) = channel();
@@ -165,7 +165,6 @@ impl<Payload: Clone> MpscTransport<Payload> {
     }
 }
 impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
-    type RecvData = Payload;
     type SendData = Payload;
 
     fn time(&mut self) -> i64 {
@@ -173,9 +172,9 @@ impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
     }
 
     fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Payload) {
-        let _ = self.channel.send(PacketType::Payload { seq_no, reply_no, payload: payload.clone() });
+        let _ = self.channel.send(PacketType::Payload(seq_no, reply_no, payload.clone() ));
     }
     fn send_ack(&mut self, reply_no: SeqNo) {
-        let _ = self.channel.send(PacketType::Ack { reply_no });
+        let _ = self.channel.send(PacketType::Ack ( reply_no ));
     }
 }

src/transport_layer.rs

@@ -7,7 +7,6 @@ use crate::SeqNo;
 /// It is possible through these generics to make SeqEx no-alloc and zero-copy, but otherwise
 /// they are most easily implemented as some combination of custom enums, `Vec<u8>` and `Arc<[u8]>`.
 pub trait TransportLayer: Clone {
-    type RecvData;
     type SendData;
 
     fn time(&mut self) -> i64;