added to API

examples/calculator.rs

@@ -4,7 +4,7 @@ use std::{
     time::Duration,
 };
 
-use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync};
+use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
 
 #[derive(Clone)]
 enum Packet {
@@ -52,9 +52,9 @@ fn receive<'a>(
                 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);
+            Ok(PacketType::Payload { seq_no, reply_no, payload }) => {
+                if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) {
+                    process(guard, packet, send_data, value);
                     true
                 } else {
                     false
@@ -63,8 +63,8 @@ fn receive<'a>(
             _ => return,
         };
         if do_pump {
-            while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) {
-                process(guard, recv_packet, send_packet, value);
+            while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) {
+                process(guard, packet, send_data, value);
             }
         }
     }

examples/hello_world.rs

@@ -1,6 +1,6 @@
 use std::sync::mpsc::Receiver;
 
-use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync};
+use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
 
 #[derive(Clone, Debug)]
 enum Packet {
@@ -45,15 +45,15 @@ fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTran
         }
         PacketType::EmptyReply { reply_no } => {
             let result = seq.receive_empty_reply(reply_no);
-            if let Some(Exclamation) = result {
+            if let Some(Exclamation) = &result {
                 // Our Hello World exchange ends right here.
                 print!("\n");
             }
             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);
+        PacketType::Payload { seq_no, reply_no, payload } => {
+            if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) {
+                process(guard, packet, send_data);
                 true
             } else {
                 false
@@ -61,8 +61,8 @@ fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTran
         }
     };
     if do_pump {
-        while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) {
-            process(guard, recv_packet, send_packet);
+        while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) {
+            process(guard, packet, send_data);
         }
     }
 }

src/seq_queue.rs

@@ -55,6 +55,7 @@ pub struct SeqEx<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN
     /// remote peer.
     /// It can be statically or dynamically set, it is up to the user to decide.
     pub resend_interval: i64,
+    pub next_service_timestamp: i64,
     next_send_seq_no: SeqNo,
     pre_recv_seq_no: SeqNo,
     send_window: [Option<SendEntry<TL>>; SLEN],
@@ -72,7 +73,7 @@ struct RecvEntry<TL: TransportLayer> {
 struct SendEntry<TL: TransportLayer> {
     seq_no: SeqNo,
     reply_no: Option<SeqNo>,
-    next_resent_time: i64,
+    next_resend_time: i64,
     data: TL::SendData,
 }
 
@@ -116,6 +117,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
     pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
         Self {
             resend_interval: retry_interval,
+            next_service_timestamp: i64::MIN,
             next_send_seq_no: initial_seq_no,
             pre_recv_seq_no: initial_seq_no.wrapping_sub(1),
             recv_window: core::array::from_fn(|_| None),
@@ -159,18 +161,25 @@ 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 try_send(&mut self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
+    pub fn try_send(&mut self, mut app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
         if self.is_full() {
             return Err(packet_data);
         }
         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 = app.time() + self.resend_interval;
-        let entry = self.send_window[seq_no as usize % self.send_window.len()].insert(SendEntry {
+        let current_time = app.time();
+        let next_resend_time = current_time + self.resend_interval;
+        if self.next_service_timestamp > next_resend_time {
+            self.next_service_timestamp = next_resend_time;
+            app.update_service_time(current_time, next_resend_time);
+        }
+        let i = seq_no as usize % self.send_window.len();
+        debug_assert!(self.send_window[i].is_none());
+        let entry = self.send_window[i].insert(SendEntry {
             seq_no,
             reply_no: None,
-            next_resent_time,
+            next_resend_time,
             data: packet_data,
         });
 
@@ -180,7 +189,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
 
     pub fn receive_raw<P: Into<TL::RecvData>>(
         &mut self,
-        app: TL,
+        mut app: TL,
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
@@ -256,7 +265,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
         let i = reply_no as usize % self.send_window.len();
         if let Some(entry) = self.send_window[i].as_mut() {
             if entry.seq_no == reply_no {
-                entry.next_resent_time = i64::MAX;
+                entry.next_resend_time = i64::MAX;
             }
         }
     }
@@ -298,41 +307,30 @@ impl<TL: TransportLayer> SeqEx<TL> {
         }
     }
 
-    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() {
-            if let Some(entry) = item {
-                if entry.next_resent_time <= current_time {
-                    entry.next_resent_time = next_interval;
-                    app.send(entry.seq_no, entry.reply_no, &entry.data);
-                } else {
-                    next_activity = next_activity.min(entry.next_resent_time);
-                }
-            }
-        }
-        next_activity - current_time
-    }
-
-    pub fn reply_raw(&mut self, app: TL, reply_no: SeqNo, packet_data: TL::SendData) {
+    pub fn reply_raw(&mut self, mut app: TL, reply_no: SeqNo, packet_data: TL::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);
 
             let i = seq_no as usize % self.send_window.len();
-            let next_resent_time = app.time() + self.resend_interval;
+            let current_time = app.time();
+            let next_resend_time = current_time + self.resend_interval;
+            if self.next_service_timestamp > next_resend_time {
+                self.next_service_timestamp = next_resend_time;
+                app.update_service_time(current_time, next_resend_time);
+            }
+            debug_assert!(self.send_window[i].is_none());
             let entry = self.send_window[i].insert(SendEntry {
                 seq_no,
                 reply_no: Some(reply_no),
-                next_resent_time,
+                next_resend_time,
                 data: packet_data,
             });
 
             app.send(entry.seq_no, entry.reply_no, &entry.data);
         }
     }
-    pub fn reply_empty_raw(&mut self, app: TL, reply_no: SeqNo) {
+    pub fn reply_empty_raw(&mut self, mut app: TL, reply_no: SeqNo) {
         if self.remove_reservation(reply_no) {
             app.send_empty_reply(reply_no);
         }
@@ -348,6 +346,23 @@ impl<TL: TransportLayer> SeqEx<TL> {
         false
     }
 
+    pub fn service(&mut self, mut app: TL) -> i64 {
+        let current_time = app.time();
+        let next_interval = current_time + self.resend_interval;
+        let mut next_activity = i64::MAX;
+        for entry in self.send_window.iter_mut().flatten() {
+            if entry.next_resend_time <= current_time {
+                entry.next_resend_time = next_interval;
+                app.send(entry.seq_no, entry.reply_no, &entry.data);
+            } else {
+                next_activity = next_activity.min(entry.next_resend_time);
+            }
+        }
+        self.next_service_timestamp = next_activity;
+        app.update_service_time(current_time, next_activity);
+        self.resend_interval.min(next_activity - current_time)
+    }
+
     pub fn iter(&self) -> Iter<'_, TL> {
         Iter(self.send_window.iter())
     }

src/single_thread.rs

@@ -19,6 +19,12 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
     }
 }
 
+pub struct RecvSuccess<'a, TL: TransportLayer, P> {
+    pub guard: ReplyGuard<'a, TL>,
+    pub packet: P,
+    pub send_data: Option<TL::SendData>,
+}
+
 impl<TL: TransportLayer> SeqEx<TL> {
     pub fn receive<P: Into<TL::RecvData>>(
         &mut self,
@@ -26,12 +32,12 @@ impl<TL: TransportLayer> SeqEx<TL> {
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
-    ) -> Result<(ReplyGuard<'_, TL>, P, Option<TL::SendData>), Error> {
+    ) -> Result<RecvSuccess<'_, TL, P>, Error> {
         self.receive_raw(app.clone(), seq_no, reply_no, packet)
-            .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
+            .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
-    pub fn pump(&mut self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option<TL::SendData>), Error> {
+    pub fn pump(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData>, Error> {
         self.pump_raw()
-            .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
+            .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
 }

src/sync.rs

@@ -35,6 +35,12 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
     }
 }
 
+pub struct RecvSuccess<'a, TL: TransportLayer, P> {
+    pub guard: ReplyGuard<'a, TL>,
+    pub packet: P,
+    pub send_data: Option<TL::SendData>,
+}
+
 impl<TL: TransportLayer> SeqExSync<TL> {
     pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
         Self {
@@ -50,17 +56,17 @@ impl<TL: TransportLayer> SeqExSync<TL> {
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         packet: P,
-    ) -> Result<(ReplyGuard<'_, TL>, P, Option<TL::SendData>), Error> {
+    ) -> Result<RecvSuccess<'_, TL, P>, Error> {
         let mut seq = self.lock();
         let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet);
         self.unblock(ret.is_ok());
-        ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
+        ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
-    pub fn pump(&self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option<TL::SendData>), Error> {
+    pub fn pump(&self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData>, Error> {
         let mut seq = self.lock();
         let ret = seq.pump_raw();
         self.unblock(ret.is_ok());
-        ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
+        ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
     }
     #[inline]
     fn unblock(&self, is_ok: bool) {
@@ -109,7 +115,7 @@ impl<TL: TransportLayer> Default for SeqExSync<TL> {
 
 #[derive(Clone)]
 pub enum PacketType<Payload: Clone> {
-    Data {
+    Payload {
         seq_no: SeqNo,
         reply_no: Option<SeqNo>,
         payload: Payload,
@@ -140,17 +146,17 @@ impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
     type RecvData = Payload;
     type SendData = Payload;
 
-    fn time(&self) -> i64 {
+    fn time(&mut self) -> i64 {
         self.time.elapsed().as_millis() as i64
     }
 
-    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(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Payload) {
+        let _ = self.channel.send(PacketType::Payload { seq_no, reply_no, payload: payload.clone() });
     }
-    fn send_ack(&self, reply_no: SeqNo) {
+    fn send_ack(&mut self, reply_no: SeqNo) {
         let _ = self.channel.send(PacketType::Ack { reply_no });
     }
-    fn send_empty_reply(&self, reply_no: SeqNo) {
+    fn send_empty_reply(&mut self, reply_no: SeqNo) {
         let _ = self.channel.send(PacketType::EmptyReply { reply_no });
     }
 }

src/transport_layer.rs

@@ -10,9 +10,11 @@ pub trait TransportLayer: Sized + Clone {
     type RecvData;
     type SendData;
 
-    fn time(&self) -> i64;
+    fn time(&mut self) -> i64;
+    #[allow(unused)]
+    fn update_service_time(&mut self, current_time: i64, timestamp: i64) {}
 
-    fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData);
-    fn send_ack(&self, reply_no: SeqNo);
-    fn send_empty_reply(&self, reply_no: SeqNo);
+    fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData);
+    fn send_ack(&mut self, reply_no: SeqNo);
+    fn send_empty_reply(&mut self, reply_no: SeqNo);
 }