rpfm_server/
session.rs

1//---------------------------------------------------------------------------//
2// Copyright (c) 2017-2026 Ismael Gutiérrez González. All rights reserved.
3//
4// This file is part of the Rusted PackFile Manager (RPFM) project,
5// which can be found here: https://github.com/Frodo45127/rpfm.
6//
7// This file is licensed under the MIT license, which can be found here:
8// https://github.com/Frodo45127/rpfm/blob/master/LICENSE.
9//---------------------------------------------------------------------------//
10
11//! Per-client session state and lifecycle.
12//!
13//! Each WebSocket connection (and each MCP client) is wrapped in a [`Session`]
14//! managed by a [`SessionManager`]. Sessions are isolated: open packs in one
15//! session aren't visible from another, and each one owns a dedicated
16//! background thread (see [`crate::background_thread`]) that processes its
17//! commands serially, with its own settings. Sessions start with the settings
18//! in the settings file, and clients owning their own settings (the UI) replace
19//! them with a `session.configure` request.
20//!
21//! ## Lifecycle
22//!
23//! 1. **Create.** A new session gets a unique [`SessionId`] from the manager
24//!    plus a fresh background thread of its own.
25//! 2. **Connect / disconnect.** Clients increment [`Session::connect`] on
26//!    attach and [`Session::disconnect`] on detach. The connection count is
27//!    what the timeout logic watches.
28//! 3. **Reconnect.** A client can pass its previous `session_id` back on the
29//!    next WebSocket handshake to adopt the same session and recover its
30//!    in-memory state. See [`SessionManager::get_or_create_session`].
31//! 4. **Timeout.** When the connection count drops to zero, the session
32//!    enters a [`DEFAULT_SESSION_TIMEOUT_SECS`]-long grace period. Reconnects
33//!    cancel the timeout; otherwise the cleanup task removes the session and
34//!    its background thread exits.
35//! 5. **Empty manager → process exit.** When a WebSocket session is removed
36//!    and no other session is left, the server process terminates, so no
37//!    orphaned backend lingers in the background. Removing the MCP session
38//!    never stops the process.
39//!
40//! ## The MCP session
41//!
42//! MCP clients share one session ([`SessionManager::mcp_session`]), whatever
43//! their protocol version: newer MCP versions have no sessions of their own,
44//! and the older ones have no disconnect signal the server can rely on. The
45//! MCP session is never counted as connected; instead, the periodic cleanup
46//! task removes it once no request has been sent through it for
47//! [`DEFAULT_SESSION_TIMEOUT_SECS`], as long as it has no open packs and no
48//! unfinished jobs. The next MCP request creates a new one.
49
50use tokio::sync::mpsc::{error::SendError, unbounded_channel, UnboundedReceiver, UnboundedSender};
51use tokio::time::{Duration, Instant};
52
53use std::collections::{BTreeMap, BTreeSet, HashMap};
54use std::sync::{Arc, Mutex, RwLock, atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering}};
55
56use rpfm_ipc::api::{ApiError, Done, Request, RpcOutcome, RpcRequest, RpcResponse};
57use rpfm_ipc::api::diagnostics::RunDiagnostics;
58use rpfm_ipc::api::jobs::JobStarted;
59use rpfm_ipc::api::session::Configure;
60use rpfm_ipc::helpers::SessionInfo;
61use rpfm_ipc::settings::Settings;
62use rpfm_ipc::settings_keys::AUTOSAVE_INTERVAL;
63use rpfm_telemetry::{error, info};
64
65use crate::api;
66use crate::background_thread;
67use crate::apply_telemetry_settings;
68use crate::jobs::{self, JobRegistry};
69
70/// Error messages for session communication.
71pub const SESSION_SENDER_ERROR: &str = "Error in session communication system. Sender failed to send message.";
72
73/// Default session timeout in seconds (5 minutes).
74pub const DEFAULT_SESSION_TIMEOUT_SECS: u64 = 300;
75
76/// How often expired sessions are looked for.
77const CLEANUP_INTERVAL: Duration = Duration::from_secs(10);
78
79//-------------------------------------------------------------------------------//
80//                              Enums & Structs
81//-------------------------------------------------------------------------------//
82
83/// Unique identifier for a session.
84pub type SessionId = u64;
85
86/// A message for a session's background thread, with where to send its response.
87#[derive(Debug)]
88pub enum SessionMessage {
89
90    /// A request.
91    Api(RpcRequest, UnboundedSender<RpcResponse>),
92
93    /// A request that runs as a job, with the ID of its job.
94    Job(u64, RpcRequest),
95
96    /// Autosaves the packs changed since their last autosave.
97    Autosave,
98
99    /// Stops the background thread.
100    Exit,
101}
102
103/// Manages all active sessions.
104///
105/// Provides thread-safe access to create, retrieve, and remove sessions.
106/// Sessions persist for a configurable timeout after all clients disconnect.
107pub struct SessionManager {
108
109    /// Map of session IDs to managed sessions.
110    sessions: Mutex<HashMap<SessionId, ManagedSession>>,
111
112    /// Counter for generating unique session IDs.
113    next_id: Mutex<SessionId>,
114
115    /// Session timeout duration.
116    timeout: Duration,
117}
118
119/// Internal state for a managed session.
120struct ManagedSession {
121
122    /// The session itself.
123    session: Arc<Session>,
124
125    /// When the last client disconnected (None if clients are connected).
126    disconnected_at: Option<Instant>,
127}
128
129/// A session represents a single client's connection state.
130///
131/// Each session has its own background thread for processing commands,
132/// ensuring complete isolation between clients.
133pub struct Session {
134
135    /// Unique identifier for this session.
136    id: SessionId,
137
138    /// Whether this is the session of the MCP clients, removed by inactivity instead of by connection count.
139    is_mcp: bool,
140
141    /// Instant of the last command sent to this session's background thread.
142    ///
143    /// Only updated by [`Session::call`], so it reflects actual work, not
144    /// transport-level pings.
145    last_activity: Mutex<Instant>,
146
147    /// Sender to communicate with this session's background thread.
148    sender: UnboundedSender<SessionMessage>,
149
150    /// Number of active connections using this session.
151    connection_count: AtomicU32,
152
153    /// Whether this session has been marked for shutdown.
154    shutdown_requested: AtomicBool,
155
156    /// Names of the packs currently open in this session, by pack key.
157    pack_names: RwLock<BTreeMap<String, String>>,
158
159    /// Jobs of this session.
160    jobs: Arc<JobRegistry>,
161
162    /// Settings the session runs with. Behind an `Arc` so requests can hold them without copying.
163    settings: RwLock<Arc<Settings>>,
164
165    /// Amount of requests sent to the background thread that it hasn't started yet.
166    queued: AtomicUsize,
167
168    /// The last diagnostics check requested, so a new one can replace it.
169    diagnostics_check: Mutex<Option<DiagnosticsCheck>>,
170}
171
172/// A diagnostics check requested to a session.
173#[derive(Debug)]
174struct DiagnosticsCheck {
175
176    /// ID of the job of the check.
177    job: u64,
178
179    /// Paths the check checks. If empty, it checks everything.
180    paths: Vec<String>,
181}
182
183//-------------------------------------------------------------------------------//
184//                             Implementations
185//-------------------------------------------------------------------------------//
186
187impl Session {
188
189    /// Create a new session with its own background thread.
190    ///
191    /// `is_mcp` marks the session of the MCP clients, which is removed by
192    /// inactivity instead of by connection count (see [`Session::is_mcp`]).
193    pub fn new(id: SessionId, is_mcp: bool) -> Arc<Self> {
194        let (sender, receiver) = unbounded_channel();
195
196        let session = Arc::new(Self {
197            id,
198            is_mcp,
199            last_activity: Mutex::new(Instant::now()),
200            sender,
201            connection_count: AtomicU32::new(0),
202            shutdown_requested: AtomicBool::new(false),
203            pack_names: RwLock::new(BTreeMap::new()),
204            jobs: Arc::new(JobRegistry::default()),
205            settings: RwLock::new(Arc::new(Settings::init(false))),
206            queued: AtomicUsize::new(0),
207            diagnostics_check: Mutex::new(None),
208        });
209
210        // Requests do heavy blocking work, so they run on their own thread instead of stalling a worker of the async runtime.
211        let session_clone = session.clone();
212        let spawned = std::thread::Builder::new().name(format!("session-{id}")).spawn(move || {
213            info!("Session {} background thread starting...", id);
214            background_thread::background_loop(receiver, session_clone);
215            info!("Session {} background thread terminated.", id);
216        });
217
218        if let Err(error) = spawned {
219            error!("Session {}: failed to start its background thread: {}", id, error);
220        }
221
222        Self::start_autosave_timer(&session);
223        session
224    }
225
226    /// Starts the timer asking the background thread to autosave the changed packs, every `autosave_interval` minutes of the settings.
227    ///
228    /// An interval of 0 disables autosaves. The timer stops when the session shuts down or is dropped.
229    fn start_autosave_timer(session: &Arc<Self>) {
230
231        // A weak reference, so the timer doesn't keep the session alive.
232        let session = Arc::downgrade(session);
233        tokio::spawn(async move {
234            while let Some(minutes) = session.upgrade().map(|session| session.settings().i32(AUTOSAVE_INTERVAL)) {
235
236                // With autosaves disabled, wait a minute and check again, in case the settings change.
237                tokio::time::sleep(Duration::from_secs(60 * minutes.max(1) as u64)).await;
238
239                let Some(alive_session) = session.upgrade().filter(|session| !session.is_shutdown_requested()) else {
240                    break;
241                };
242
243                if minutes > 0 {
244                    let _ = alive_session.enqueue(SessionMessage::Autosave);
245                }
246            }
247        });
248    }
249
250    /// Get the session ID.
251    pub fn id(&self) -> SessionId {
252        self.id
253    }
254
255    /// Whether this is the session of the MCP clients.
256    pub fn is_mcp(&self) -> bool {
257        self.is_mcp
258    }
259
260    /// Record that a command was just sent to this session's background thread.
261    fn touch(&self) {
262        *self.last_activity.lock().unwrap() = Instant::now();
263    }
264
265    /// Get the instant of the last command sent to this session.
266    pub fn last_activity(&self) -> Instant {
267        *self.last_activity.lock().unwrap()
268    }
269
270    /// Increment the connection count.
271    pub fn connect(&self) {
272        self.connection_count.fetch_add(1, Ordering::SeqCst);
273    }
274
275    /// Decrement the connection count.
276    pub fn disconnect(&self) {
277        self.connection_count.fetch_sub(1, Ordering::SeqCst);
278    }
279
280    /// Get the current connection count.
281    pub fn connection_count(&self) -> u32 {
282        self.connection_count.load(Ordering::SeqCst)
283    }
284
285    /// Check if shutdown has been requested.
286    pub fn is_shutdown_requested(&self) -> bool {
287        self.shutdown_requested.load(Ordering::SeqCst)
288    }
289
290    /// Get the names of the packs open in this session.
291    pub fn pack_names(&self) -> Vec<String> {
292        self.pack_names.read().unwrap().values().cloned().collect()
293    }
294
295    /// Set the name of an open pack of this session, adding it if it's not there yet.
296    pub fn set_pack_name(&self, pack_key: &str, name: &str) {
297        self.pack_names.write().unwrap().insert(pack_key.to_owned(), name.to_owned());
298    }
299
300    /// Remove a pack from the open ones of this session.
301    pub fn remove_pack(&self, pack_key: &str) {
302        self.pack_names.write().unwrap().remove(pack_key);
303    }
304
305    /// Shutdown this session by stopping its background thread.
306    pub fn shutdown(&self) {
307        info!("Session {} shutting down...", self.id);
308
309        if self.shutdown_requested.swap(true, Ordering::SeqCst) {
310            info!("Session {} already marked for shutdown before...", self.id);
311            return;
312        }
313
314        // Ignore errors, as the channel may be already closed.
315        let _ = self.sender.send(SessionMessage::Exit);
316    }
317
318    /// Returns the jobs of this session.
319    pub fn jobs(&self) -> &Arc<JobRegistry> {
320        &self.jobs
321    }
322
323    /// Returns the settings of this session.
324    pub fn settings(&self) -> Arc<Settings> {
325        self.settings.read().unwrap().clone()
326    }
327
328    /// Replaces the settings of this session, applying their telemetry choices to the whole process.
329    fn configure(&self, request: RpcRequest) -> RpcResponse {
330        let result = api::parse_params::<Configure>(request.params).and_then(|configure| {
331            apply_telemetry_settings(&configure.settings);
332            *self.settings.write().unwrap() = Arc::new(configure.settings);
333            serde_json::to_value(Done {}).map_err(|error| ApiError::Internal(error.to_string()))
334        });
335
336        RpcResponse::new(request.id, result)
337    }
338
339    /// Send a request to this session's background thread.
340    ///
341    /// Configuration and job control requests are answered without waiting for the background thread,
342    /// and jobs are answered right away with their ID, before they run.
343    ///
344    /// Returns a receiver to get the response.
345    pub fn call(&self, request: RpcRequest) -> UnboundedReceiver<RpcResponse> {
346        self.touch();
347        let (sender_back, receiver_back) = unbounded_channel();
348
349        if request.method == Configure::METHOD {
350            let _ = sender_back.send(self.configure(request));
351            return receiver_back;
352        }
353
354        if jobs::is_job_control_method(&request.method) {
355            jobs::handle_request(self.jobs.clone(), request, sender_back);
356            return receiver_back;
357        }
358
359        if api::is_stateless_method(&request.method) {
360            let settings = self.settings();
361            tokio::spawn(async move {
362                let id = request.id;
363                let response = tokio::task::spawn_blocking(move || api::dispatch_stateless(request, &settings)).await
364                    .unwrap_or_else(|error| RpcResponse::new(id, Err(ApiError::Internal(format!("The background task failed: {error}")))));
365
366                let _ = sender_back.send(response);
367            });
368            return receiver_back;
369        }
370
371        if api::is_job_method(&request.method) {
372            let job = self.jobs.create(&request.method);
373            let id = request.id;
374            let request = if request.method == RunDiagnostics::METHOD {
375                self.replace_diagnostics_check(job, request)
376            } else {
377                request
378            };
379
380            if let Err(error) = self.enqueue(SessionMessage::Job(job, request)) {
381                let message = format!("{SESSION_SENDER_ERROR}: {error}");
382                info!("{message}");
383                self.jobs.finish(job, RpcOutcome::Error(ApiError::Internal(message).into()));
384            }
385
386            let started = serde_json::to_value(JobStarted { job }).map_err(|error| ApiError::Internal(error.to_string()));
387            let _ = sender_back.send(RpcResponse::new(id, started));
388            return receiver_back;
389        }
390
391        if let Err(error) = self.enqueue(SessionMessage::Api(request, sender_back)) {
392            let message = format!("{SESSION_SENDER_ERROR}: {error}");
393            info!("{message}");
394            if let SessionMessage::Api(request, sender_back) = error.0 {
395                let _ = sender_back.send(RpcResponse::new(request.id, Err(ApiError::Internal(message))));
396            }
397        }
398        receiver_back
399    }
400
401    /// Sends a request to the background thread, counting it as queued until the thread starts it.
402    fn enqueue(&self, message: SessionMessage) -> Result<(), SendError<SessionMessage>> {
403        self.queued.fetch_add(1, Ordering::SeqCst);
404        self.sender.send(message).inspect_err(|_| { self.queued.fetch_sub(1, Ordering::SeqCst); })
405    }
406
407    /// Marks a request sent to the background thread as started.
408    pub fn dequeued(&self) {
409        self.queued.fetch_sub(1, Ordering::SeqCst);
410    }
411
412    /// Returns if there are requests waiting for the background thread.
413    pub fn has_queued_requests(&self) -> bool {
414        self.queued.load(Ordering::SeqCst) > 0
415    }
416
417    /// Makes a new diagnostics check replace the last one, if it hasn't ended yet.
418    ///
419    /// The new check also checks the paths the replaced one would have checked. A queued replaced check is
420    /// cancelled, and a running one is cancelled once it stops for the new one (see [`Self::requeue_diagnostics_check`]).
421    ///
422    /// # Arguments
423    ///
424    /// * `job` - Job of the new check.
425    /// * `request` - Request of the new check.
426    ///
427    /// # Returns
428    ///
429    /// The request of the new check, with the paths of the replaced one added.
430    fn replace_diagnostics_check(&self, job: u64, mut request: RpcRequest) -> RpcRequest {
431        let Ok(mut params) = api::parse_params::<RunDiagnostics>(request.params.clone()) else {
432            return request;
433        };
434
435        let mut last_check = self.diagnostics_check.lock().unwrap();
436        let pending = last_check.take().filter(|check| self.jobs.status(check.job).is_some_and(|status| !status.state.has_ended()));
437        if let Some(pending) = pending {
438
439            // No paths means everything, so a full check covers any other.
440            params.paths = if pending.paths.is_empty() || params.paths.is_empty() {
441                vec![]
442            } else {
443                pending.paths.into_iter().chain(params.paths).collect::<BTreeSet<_>>().into_iter().collect()
444            };
445
446            let _ = self.jobs.cancel(pending.job);
447            if let Ok(params) = serde_json::to_value(&params) {
448                request.params = params;
449            }
450        }
451
452        *last_check = Some(DiagnosticsCheck { job, paths: params.paths });
453        request
454    }
455
456    /// Queues again a diagnostics check that stopped to let other requests run, or cancels it if a newer one replaced it.
457    ///
458    /// # Arguments
459    ///
460    /// * `job` - Job of the check.
461    /// * `request` - Request of the check.
462    pub fn requeue_diagnostics_check(&self, job: u64, request: RpcRequest) {
463        let last_check = self.diagnostics_check.lock().unwrap();
464        if last_check.as_ref().is_some_and(|check| check.job == job) && self.jobs.requeue(job) {
465            if self.enqueue(SessionMessage::Job(job, request)).is_err() {
466                self.jobs.finish_cancelled(job);
467            }
468        } else {
469            self.jobs.finish_cancelled(job);
470        }
471    }
472}
473
474impl Default for SessionManager {
475    fn default() -> Self {
476        Self {
477            sessions: Mutex::new(HashMap::new()),
478            next_id: Mutex::new(1),
479            timeout: Duration::from_secs(DEFAULT_SESSION_TIMEOUT_SECS),
480        }
481    }
482}
483
484impl SessionManager {
485
486    /// Create a new session and return a reference to it.
487    pub fn create_session(&self) -> Arc<Session> {
488        self.create_session_internal(false)
489    }
490
491    /// Returns the session shared by all MCP clients, creating it if there is none.
492    ///
493    /// It's never counted as connected. Instead, [`SessionManager::cleanup_expired_sessions`] removes it
494    /// once no request has been sent through it for [`DEFAULT_SESSION_TIMEOUT_SECS`], if it has no open
495    /// packs and no unfinished jobs.
496    pub fn mcp_session(&self) -> Arc<Session> {
497        let mut sessions = self.sessions.lock().unwrap();
498        let existing = sessions.values()
499            .find(|managed| managed.session.is_mcp() && !managed.session.is_shutdown_requested())
500            .map(|managed| managed.session.clone());
501
502        let session = existing.unwrap_or_else(|| self.insert_session(&mut sessions, true));
503
504        // Getting it counts as activity, so it isn't removed before the request that got it is sent.
505        session.touch();
506        session
507    }
508
509    fn create_session_internal(&self, is_mcp: bool) -> Arc<Session> {
510        let mut sessions = self.sessions.lock().unwrap();
511        self.insert_session(&mut sessions, is_mcp)
512    }
513
514    /// Creates a session and adds it to the provided sessions, which must be the locked [`SessionManager::sessions`].
515    fn insert_session(&self, sessions: &mut HashMap<SessionId, ManagedSession>, is_mcp: bool) -> Arc<Session> {
516        let id = {
517            let mut next_id = self.next_id.lock().unwrap();
518            let id = *next_id;
519            *next_id += 1;
520            id
521        };
522
523        let session = Session::new(id, is_mcp);
524
525        // The MCP session starts disconnected: its lifetime is governed by
526        // inactivity, not by the connection count.
527        if !is_mcp {
528            session.connect();
529        }
530
531        sessions.insert(id, ManagedSession {
532            session: session.clone(),
533            disconnected_at: None,
534        });
535
536        info!("Created new {} session with ID: {}", if is_mcp { "MCP" } else { "WebSocket" }, id);
537        session
538    }
539
540    /// Get an existing session by ID, or create a new one if the ID doesn't exist.
541    ///
542    /// If `session_id` is `Some`, attempts to retrieve that session.
543    /// If the session doesn't exist or `session_id` is `None`, creates a new session.
544    ///
545    /// Returns the session and whether it was newly created.
546    pub fn get_or_create_session(&self, session_id: Option<SessionId>) -> (Arc<Session>, bool) {
547        if let Some(id) = session_id {
548            let mut sessions = self.sessions.lock().unwrap();
549            if let Some(managed) = sessions.get_mut(&id) {
550
551                // Check if the session is still valid (not shut down).
552                if !managed.session.is_shutdown_requested() {
553                    managed.session.connect();
554                    managed.disconnected_at = None;
555                    info!("Client reconnected to existing session {}", id);
556                    return (managed.session.clone(), false);
557                }
558            }
559        }
560
561        // Either no session_id provided, or session not found/invalid.
562        // Create a new session.
563        (self.create_session(), true)
564    }
565
566    /// Get a session by ID without incrementing the connection count.
567    pub fn get_session(&self, id: SessionId) -> Option<Arc<Session>> {
568        let sessions = self.sessions.lock().unwrap();
569        sessions.get(&id).map(|m| m.session.clone())
570    }
571
572    /// Mark a session as disconnected by a client.
573    ///
574    /// If no more clients are connected, starts the timeout countdown.
575    /// [`SessionManager::cleanup_expired_sessions`] removes the session after the timeout unless a client reconnects.
576    pub fn client_disconnected(&self, id: SessionId) {
577        let mut sessions = self.sessions.lock().unwrap();
578        if let Some(managed) = sessions.get_mut(&id) {
579            managed.session.disconnect();
580
581            if managed.session.connection_count() == 0 {
582                managed.disconnected_at = Some(Instant::now());
583                info!("Session {} has no active connections, will timeout in {:?}", id, self.timeout);
584            }
585        }
586    }
587
588    /// Perform cleanup of expired sessions.
589    ///
590    /// This should be called periodically.
591    ///
592    /// # Returns
593    ///
594    /// If a WebSocket session was removed.
595    pub fn cleanup_expired_sessions(&self) -> bool {
596        let now = Instant::now();
597        let mut websocket_session_removed = false;
598
599        // Checked and removed under the same lock, so no client can get a session while it's being removed.
600        let mut sessions = self.sessions.lock().unwrap();
601        sessions.retain(|id, managed| {
602            let expired = if managed.session.is_mcp() {
603
604                // MCP clients have no disconnect signal, so their session is removed by inactivity,
605                // unless removing it would lose work.
606                now.duration_since(managed.session.last_activity()) >= self.timeout
607                    && managed.session.pack_names().is_empty()
608                    && !managed.session.jobs().has_unfinished_jobs()
609            } else {
610                managed.disconnected_at.is_some_and(|disconnected_at| now.duration_since(disconnected_at) >= self.timeout)
611                    && managed.session.connection_count() == 0
612            };
613
614            if expired {
615                info!("Removing session {}", id);
616                managed.session.shutdown();
617                websocket_session_removed |= !managed.session.is_mcp();
618            }
619
620            !expired
621        });
622
623        websocket_session_removed
624    }
625
626    /// Stops the server process if there are no sessions left.
627    ///
628    /// Called after a WebSocket session is removed. Removing the MCP session never stops the process,
629    /// so a server only used by MCP clients keeps running.
630    pub fn exit_if_no_sessions(&self) {
631        if self.session_count() == 0 {
632            info!("No more active sessions, shutting down server...");
633            rpfm_telemetry::flush("Server Action Telemetry");
634            std::process::exit(0);
635        }
636    }
637
638    /// Remove a session immediately.
639    pub fn remove_session(&self, id: SessionId) -> Option<Arc<Session>> {
640        let mut sessions = self.sessions.lock().unwrap();
641        if let Some(managed) = sessions.remove(&id) {
642            info!("Removing session {}", id);
643            managed.session.shutdown();
644            Some(managed.session)
645        } else {
646            None
647        }
648    }
649
650    /// Get the number of active sessions.
651    pub fn session_count(&self) -> usize {
652        let sessions = self.sessions.lock().unwrap();
653        sessions.len()
654    }
655
656    /// Get all active session IDs.
657    pub fn session_ids(&self) -> Vec<SessionId> {
658        let sessions = self.sessions.lock().unwrap();
659        sessions.keys().cloned().collect()
660    }
661
662    /// Get information about all active sessions.
663    ///
664    /// Returns a vector of [`SessionInfo`] structs containing session state snapshots
665    /// for use by session management tools.
666    pub fn get_sessions_info(&self) -> Vec<SessionInfo> {
667        let sessions = self.sessions.lock().unwrap();
668        let now = Instant::now();
669
670        sessions.values().map(|managed| {
671            let timeout_remaining_secs = managed.disconnected_at.map(|disconnected_at| {
672                let elapsed = now.duration_since(disconnected_at);
673                if elapsed < self.timeout {
674                    (self.timeout - elapsed).as_secs()
675                } else {
676                    0
677                }
678            });
679
680            SessionInfo::new(
681                managed.session.id(),
682                managed.session.connection_count(),
683                timeout_remaining_secs,
684                managed.session.is_shutdown_requested(),
685                managed.session.pack_names(),
686            )
687        }).collect()
688    }
689
690    /// Start a background task that periodically cleans up expired sessions, and stops the server when none are left.
691    pub fn start_cleanup_task(manager: Arc<Self>) {
692        tokio::spawn(async move {
693            loop {
694                tokio::time::sleep(CLEANUP_INTERVAL).await;
695                if manager.cleanup_expired_sessions() {
696                    manager.exit_if_no_sessions();
697                }
698            }
699        });
700    }
701}
702
703#[cfg(test)]
704#[path = "session_test.rs"]
705mod session_test;
rpfm_server/session.rs

rpfm_server/
session.rs
