1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
use crate::client::{INextClientInner, NetXClient, SessionSave};
use aqueue::Actor;
use std::collections::VecDeque;
use std::sync::{Arc, Weak};
use std::time::{Duration, Instant};
use tokio::time::sleep;
/// `RequestManager` manages the requests, including their timeouts and the associated client.
pub struct RequestManager<T> {
queue: VecDeque<(i64, Instant)>,
request_out_time: u32,
netx_client: Weak<Actor<NetXClient<T>>>,
}
unsafe impl<T> Send for RequestManager<T> {}
unsafe impl<T> Sync for RequestManager<T> {}
impl<T> Drop for RequestManager<T> {
/// Custom drop implementation for `RequestManager` to log when it is dropped.
fn drop(&mut self) {
log::debug!("request manager is drop");
}
}
impl<T: SessionSave + 'static> RequestManager<T> {
/// Creates a new `RequestManager` instance.
///
/// # Arguments
///
/// * `request_out_time` - The timeout duration for requests.
/// * `netx_client` - A weak reference to the `NetXClient`.
///
/// # Returns
///
/// An `Arc` wrapped `Actor` of `RequestManager`.
pub fn new(
request_out_time: u32,
netx_client: Weak<Actor<NetXClient<T>>>,
) -> Arc<Actor<RequestManager<T>>> {
let ptr = Arc::new(Actor::new(RequestManager {
queue: VecDeque::new(),
request_out_time,
netx_client,
}));
Self::start_check(Arc::downgrade(&ptr));
ptr
}
/// Starts the periodic check for request timeouts.
///
/// # Arguments
///
/// * `request_manager` - A weak reference to the `RequestManager`.
fn start_check(request_manager: Weak<Actor<RequestManager<T>>>) {
tokio::spawn(async move {
while let Some(req) = request_manager.upgrade() {
if let Err(er) = req.check().await {
log::error!("check request error:{}", er);
}
// Drop the strong reference before sleeping so the RequestManager
// can be freed promptly when the client is dropped, rather than
// being kept alive for the full 500 ms sleep window.
drop(req);
sleep(Duration::from_millis(500)).await
}
});
}
/// Checks the queue for timed-out requests and handles them.
#[inline]
pub async fn check(&mut self) {
// Upgrade once here rather than inside the loop — if multiple entries
// have timed out simultaneously we avoid redundant Weak::upgrade() calls.
let client = self.netx_client.upgrade();
while let Some(item) = self.queue.pop_back() {
if item.1.elapsed().as_millis() as u32 >= self.request_out_time {
if let Some(ref c) = client {
c.set_error(item.0, crate::error::Error::SerialTimeOut(item.0))
.await;
}
} else {
self.queue.push_back(item);
break;
}
}
}
/// Adds a new request to the queue with the current timestamp.
///
/// # Arguments
///
/// * `session_id` - The ID of the session to be added.
#[inline]
pub fn set(&mut self, session_id: i64) {
self.queue.push_front((session_id, Instant::now()));
}
}
/// Trait defining the interface for `RequestManager`.
pub(crate) trait IRequestManager {
/// Asynchronously checks the requests.
async fn check(&self) -> crate::error::Result<()>;
/// Asynchronously sets a new request.
///
/// # Arguments
///
/// * `session_id` - The ID of the session to be set.
async fn set(&self, session_id: i64) -> crate::error::Result<()>;
}
impl<T: SessionSave + 'static> IRequestManager for Actor<RequestManager<T>> {
/// Asynchronously checks the requests by calling the inner `check` method.
#[inline]
async fn check(&self) -> crate::error::Result<()> {
self.inner_call(|inner| async move {
inner.get_mut().check().await;
Ok(())
})
.await
}
/// Asynchronously sets a new request by calling the inner `set` method.
///
/// # Arguments
///
/// * `session_id` - The ID of the session to be set.
#[inline]
async fn set(&self, session_id: i64) -> crate::error::Result<()> {
self.inner_call(|inner| async move {
inner.get_mut().set(session_id);
Ok(())
})
.await
}
}