bump version to 0.2.86 and refactor new timer

Author: shenjinti

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsipstack"
-version = "0.2.85"
+version = "0.2.86"
 edition = "2021"
 description = "SIP Stack Rust library for building SIP applications"
 license = "MIT"

src/dialog/dialog.rs

@@ -184,45 +184,6 @@ pub struct DialogInner {
     pub(super) initial_request: Request,
 }
 
-// pub struct DialogStateReceiver {
-//     pub(super) dialog_layer_inner: DialogLayerInnerRef,
-//     pub(super) receiver: UnboundedReceiver<DialogState>,
-//     pub(super) dialog_id: Option<DialogId>,
-// }
-
-// impl DialogStateReceiver {
-//     pub async fn recv(&mut self) -> Option<DialogState> {
-//         let state = self.receiver.recv().await;
-//         if let Some(ref s) = state {
-//             self.dialog_id = Some(s.id().clone());
-//         }
-//         state
-//     }
-// }
-
-// impl Drop for DialogStateReceiver {
-//     fn drop(&mut self) {
-//         let id = match self.dialog_id.take() {
-//             Some(id) => id,
-//             None => return,
-//         };
-
-//         match self.dialog_layer_inner.dialogs.write().as_mut() {
-//             Ok(dialogs) => {
-//                 if let Some(dialog) = dialogs.remove(&id) {
-//                     info!(%id, "dialog removed on state receiver drop");
-//                     tokio::spawn(async move {
-//                         if let Err(e) = dialog.hangup().await {
-//                             warn!(%id, "error hanging up dialog on drop: {}", e);
-//                         }
-//                     });
-//                 }
-//             }
-//             Err(_) => {}
-//         }
-//     }
-// }
-
 pub type DialogStateReceiver = UnboundedReceiver<DialogState>;
 pub type DialogStateSender = UnboundedSender<DialogState>;
 

src/transaction/endpoint.rs

@@ -15,7 +15,7 @@ use rsip::{prelude::HeadersExt, SipMessage};
 use std::{
     collections::HashMap,
     sync::{Arc, Mutex, RwLock},
-    time::{Duration, Instant},
+    time::Duration,
 };
 use tokio::{
     select,
@@ -108,6 +108,7 @@ pub struct EndpointInner {
     incoming_sender: TransactionSender,
     incoming_receiver: Mutex<Option<TransactionReceiver>>,
     cancel_token: CancellationToken,
+    #[allow(dead_code)]
     timer_interval: Duration,
     pub(super) message_inspector: Option<Box<dyn MessageInspector>>,
     pub(super) locator: Option<Box<dyn TargetLocator>>,
@@ -235,13 +236,10 @@ impl EndpointInner {
 
     pub async fn serve(self: &Arc<Self>) -> Result<()> {
         select! {
-            _ = self.cancel_token.cancelled() => {
-            },
-            r = self.process_timer() => {
-                _ = r?
-            },
+            _ = self.cancel_token.cancelled() => {},
+            _ = self.process_timer() => {},
             r = self.clone().process_transport_layer() => {
-                _ = r?
+                _ = r?;
             },
         }
         Ok(())
@@ -290,10 +288,9 @@ impl EndpointInner {
         Ok(())
     }
 
-    pub async fn process_timer(&self) -> Result<()> {
-        let mut ticker = tokio::time::interval(self.timer_interval);
+    pub async fn process_timer(&self) {
         loop {
-            for t in self.timers.poll(Instant::now()) {
+            for t in self.timers.wait_for_ready().await.into_iter() {
                 match t {
                     TransactionTimer::TimerCleanup(key) => {
                         trace!(%key, "TimerCleanup");
@@ -326,7 +323,6 @@ impl EndpointInner {
                     }
                 }
             }
-            ticker.tick().await;
         }
     }