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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
use crate::cleanup_diag::{self, CleanupStage};
use crate::state_table::{CallId, DialogId, MediaSessionId, SessionId};
use dashmap::DashMap;
use std::sync::Arc;
use tracing::{debug, info};
use super::state::SessionState;
use crate::state_table::Role;
use crate::types::CallState;
/// Flexible session storage for rvoip-sip
///
/// This store supports multiple sessions for legitimate use cases like transfers,
/// while keeping the API simple for single session usage.
///
/// Uses DashMap for lock-free concurrent access to avoid deadlocks.
pub struct SessionStore {
/// Multiple session storage (lock-free with DashMap)
pub(crate) sessions: Arc<DashMap<SessionId, SessionState>>,
/// Index by dialog ID (lock-free with DashMap)
pub(crate) by_dialog: Arc<DashMap<DialogId, SessionId>>,
/// Index by call ID (lock-free with DashMap)
pub(crate) by_call_id: Arc<DashMap<CallId, SessionId>>,
/// Index by media session ID (lock-free with DashMap)
pub(crate) by_media_id: Arc<DashMap<MediaSessionId, SessionId>>,
}
impl SessionStore {
/// Create a new session store
pub fn new() -> Self {
Self {
sessions: Arc::new(DashMap::new()),
by_dialog: Arc::new(DashMap::new()),
by_call_id: Arc::new(DashMap::new()),
by_media_id: Arc::new(DashMap::new()),
}
}
/// Create a new session store with preallocated index capacity.
pub fn with_capacity(capacity: usize) -> Self {
Self {
sessions: Arc::new(DashMap::with_capacity(capacity)),
by_dialog: Arc::new(DashMap::with_capacity(capacity)),
by_call_id: Arc::new(DashMap::with_capacity(capacity)),
by_media_id: Arc::new(DashMap::with_capacity(capacity)),
}
}
/// Create a new session (supports multiple sessions for transfers)
pub async fn create_session(
&self,
session_id: SessionId,
role: Role,
with_history: bool,
) -> Result<SessionState, Box<dyn std::error::Error + Send + Sync>> {
let session = if with_history {
use crate::session_store::HistoryConfig;
SessionState::with_history(session_id.clone(), role, HistoryConfig::default())
} else {
SessionState::new(session_id.clone(), role)
};
// DashMap: Lock-free insert with check
if self.sessions.contains_key(&session_id) {
return Err(format!("Session {} already exists", session_id).into());
}
self.sessions.insert(session_id.clone(), session.clone());
info!("Created new session {} with role {:?}", session_id, role);
Ok(session)
}
/// Get a session by ID
pub async fn get_session(
&self,
session_id: &SessionId,
) -> Result<SessionState, Box<dyn std::error::Error + Send + Sync>> {
// DashMap: Lock-free read
debug!(
"Looking for session {}, store has {} sessions",
session_id,
self.sessions.len()
);
self.sessions
.get(session_id)
.map(|entry| entry.value().clone())
.ok_or_else(|| format!("Session {} not found", session_id).into())
}
/// Update a session
pub async fn update_session(
&self,
session: SessionState,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let session_id = session.session_id.clone();
// DashMap: Lock-free update with index management
// Get old session to check for index changes
if let Some(old_session) = self
.sessions
.get(&session_id)
.map(|entry| entry.value().clone())
{
// Update indexes if IDs have changed
if old_session.dialog_id != session.dialog_id {
if let Some(old_id) = &old_session.dialog_id {
self.by_dialog.remove(old_id);
}
if let Some(new_id) = &session.dialog_id {
self.by_dialog.insert(new_id.clone(), session_id.clone());
}
}
if old_session.media_session_id != session.media_session_id {
if let Some(old_id) = &old_session.media_session_id {
self.by_media_id.remove(old_id);
}
if let Some(new_id) = &session.media_session_id {
self.by_media_id.insert(new_id.clone(), session_id.clone());
}
}
if old_session.call_id != session.call_id {
if let Some(old_id) = &old_session.call_id {
self.by_call_id.remove(old_id);
}
if let Some(new_id) = &session.call_id {
self.by_call_id.insert(new_id.clone(), session_id.clone());
}
}
}
self.sessions.insert(session_id.clone(), session);
debug!("Updated session {}", session_id);
Ok(())
}
/// Remove a session
pub async fn remove_session(
&self,
session_id: &SessionId,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let guard = cleanup_diag::stage_guard(CleanupStage::SessionStoreRemoval, &session_id.0);
// DashMap: Lock-free remove
if let Some((_, session)) = self.sessions.remove(session_id) {
// Clean up indexes
if let Some(dialog_id) = &session.dialog_id {
self.by_dialog.remove(dialog_id);
}
if let Some(media_id) = &session.media_session_id {
self.by_media_id.remove(media_id);
}
if let Some(call_id) = &session.call_id {
self.by_call_id.remove(call_id);
}
info!("Removed session {}", session_id);
guard.finish_success();
Ok(())
} else {
Err(format!("Session {} not found", session_id).into())
}
}
/// Find session by dialog ID
pub async fn find_by_dialog(&self, dialog_id: &DialogId) -> Option<SessionState> {
// DashMap: Lock-free lookup
self.by_dialog.get(dialog_id).and_then(|entry| {
let session_id = entry.value();
self.sessions.get(session_id).map(|s| s.value().clone())
})
}
/// Find session by media session ID
pub async fn find_by_media_id(&self, media_id: &MediaSessionId) -> Option<SessionState> {
// DashMap: Lock-free lookup
self.by_media_id.get(media_id).and_then(|entry| {
let session_id = entry.value();
self.sessions.get(session_id).map(|s| s.value().clone())
})
}
/// Find session by call ID
pub async fn find_by_call_id(&self, call_id: &CallId) -> Option<SessionState> {
// DashMap: Lock-free lookup
self.by_call_id.get(call_id).and_then(|entry| {
let session_id = entry.value();
self.sessions.get(session_id).map(|s| s.value().clone())
})
}
/// Get all active sessions
pub async fn get_all_sessions(&self) -> Vec<SessionState> {
// DashMap: Lock-free iteration
self.sessions
.iter()
.map(|entry| entry.value().clone())
.collect()
}
// ===== Multi-Session Utility Methods =====
/// Check if any sessions exist
pub async fn has_session(&self) -> bool {
// DashMap: Lock-free check
!self.sessions.is_empty()
}
/// Get the most recent session ID (for API compatibility)
pub async fn get_current_session_id(&self) -> Option<SessionId> {
// DashMap: Lock-free iteration
self.sessions
.iter()
.max_by_key(|entry| entry.value().created_at)
.map(|entry| entry.key().clone())
}
/// Clear all session data (complete reset)
pub async fn clear(&self) {
// DashMap: Lock-free clear
self.sessions.clear();
self.by_dialog.clear();
self.by_call_id.clear();
self.by_media_id.clear();
info!("Cleared all session data");
}
/* Old cleanup - replaced by cleanup.rs
/// Clean up stale sessions
pub async fn cleanup_stale_sessions_old(&self, max_age: Duration) {
let mut sessions = self.sessions.write().await;
let now = Instant::now();
let mut to_remove = Vec::new();
for (id, session) in sessions.iter() {
let should_remove = match session.call_state {
CallState::Terminated | CallState::Failed(_) => {
// Keep terminated sessions for a short time
session.time_in_state() > Duration::from_secs(300)
}
_ => {
// Remove long-idle sessions
session.session_duration() > max_age
}
};
if should_remove {
to_remove.push(id.clone());
}
}
for id in to_remove {
if let Some(session) = sessions.remove(&id) {
// Clean up indexes
if let Some(dialog_id) = &session.dialog_id {
self.by_dialog.write().await.remove(dialog_id);
}
if let Some(media_id) = &session.media_session_id {
self.by_media_id.write().await.remove(media_id);
}
if let Some(call_id) = &session.call_id {
self.by_call_id.write().await.remove(call_id);
}
warn!("Cleaned up stale session {}", id);
}
}
}
*/
/// Get session statistics
pub async fn get_stats(&self) -> SessionStats {
// DashMap: Lock-free iteration
let mut stats = SessionStats::default();
for entry in self.sessions.iter() {
let session = entry.value();
stats.total += 1;
match session.call_state {
CallState::Idle => stats.idle += 1,
CallState::Initiating => stats.initiating += 1,
CallState::CancelPending => stats.initiating += 1,
CallState::Cancelling => stats.terminating += 1,
CallState::Ringing => stats.ringing += 1,
CallState::Answering => stats.ringing += 1, // Still in setup phase
CallState::AnsweringHangupPending => stats.terminating += 1,
CallState::EarlyMedia => stats.active += 1, // Count early media as active
CallState::Active => stats.active += 1,
CallState::HoldPending => stats.active += 1, // Count hold-pending as active until 2xx commits
CallState::OnHold => stats.on_hold += 1,
CallState::Resuming => stats.active += 1, // Count resuming as active
CallState::Bridged => stats.active += 1, // Count bridged as active
CallState::Transferring => stats.active += 1, // Count transferring as active
CallState::TransferringCall => stats.active += 1, // Count transfer recipient as active
CallState::Terminating => stats.terminating += 1,
CallState::Terminated => stats.terminated += 1,
CallState::Cancelled => stats.terminated += 1, // Count cancelled as terminated
CallState::Failed(_) => stats.failed += 1,
CallState::Muted => stats.active += 1, // Count as active
CallState::ConsultationCall => stats.active += 1, // Count as active
// Registration states
CallState::Registering => stats.initiating += 1, // Count as initiating
CallState::Registered => stats.idle += 1, // Count as idle (ready for calls)
CallState::Unregistering => stats.terminating += 1, // Count as terminating
// Subscription/Presence states
CallState::Subscribing => stats.initiating += 1, // Count as initiating
CallState::Subscribed => stats.idle += 1, // Count as idle (subscription active)
CallState::Publishing => stats.initiating += 1, // Count as initiating
// Authentication and routing states
CallState::Authenticating => stats.initiating += 1, // Count as initiating
CallState::Messaging => stats.active += 1, // Count as active
}
}
stats
}
}
/// Session statistics
#[derive(Debug, Default, Clone)]
pub struct SessionStats {
pub total: usize,
pub idle: usize,
pub initiating: usize,
pub ringing: usize,
pub active: usize,
pub on_hold: usize,
pub terminating: usize,
pub terminated: usize,
pub failed: usize,
}