infrarust 1.6.1

A Rust universal Minecraft proxy
Documentation 
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

use infrarust_config::LogType;
use tokio::task::JoinHandle;
use tracing::{debug, info};

use super::ActorSupervisor;

impl ActorSupervisor {
    pub async fn shutdown_actors(&self, config_id: &str) {
        let mut actors = self.actors.write().await;
        if let Some(pairs) = actors.get_mut(config_id) {
            for pair in pairs.iter() {
                debug!(
                    log_type = LogType::Supervisor.as_str(),
                    "Shutting down actor for user {}", pair.username
                );
                pair.shutdown
                    .store(true, std::sync::atomic::Ordering::SeqCst);
            }

            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            pairs.clear();
        }

        let mut tasks = self.tasks.write().await;
        if let Some(task_handles) = tasks.remove(config_id) {
            for handle in task_handles {
                handle.abort();
            }
        }
    }

    pub async fn register_task(&self, config_id: &str, handle: JoinHandle<()>) {
        let mut tasks = self.tasks.write().await;
        tasks
            .entry(config_id.to_string())
            .or_insert_with(Vec::new)
            .push(handle);
    }

    pub async fn health_check(&self) {
        let mut actors = self.actors.write().await;
        let mut tasks = self.tasks.write().await;

        for (config_id, pairs) in actors.iter_mut() {
            let before_count = pairs.len();

            // Log any player disconnections before removing them
            for pair in pairs.iter() {
                if pair.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
                    self.log_disconnect_if_needed(pair).await;
                }
            }

            // Remove actors with shutdown flag set
            pairs.retain(|pair| !pair.shutdown.load(std::sync::atomic::Ordering::SeqCst));

            let after_count = pairs.len();
            if before_count != after_count {
                debug!(
                    log_type = LogType::Supervisor.as_str(),
                    "Cleaned up {} dead actors for config {}",
                    before_count - after_count,
                    config_id
                );

                // Clean up any associated tasks
                if let Some(task_handles) = tasks.get_mut(config_id) {
                    while task_handles.len() > pairs.len() {
                        if let Some(handle) = task_handles.pop() {
                            debug!(
                                log_type = LogType::Supervisor.as_str(),
                                "Aborting orphaned task for {}", config_id
                            );
                            handle.abort();
                        }
                    }
                }
            }
        }

        // Check for stale tasks without associated actors
        tasks.retain(|config_id, handles| {
            if !actors.contains_key(config_id) || actors[config_id].is_empty() {
                for handle in handles.iter() {
                    debug!(
                        log_type = LogType::Supervisor.as_str(),
                        "Aborting orphaned task for {}", config_id
                    );
                    handle.abort();
                }
                false
            } else {
                true
            }
        });
    }

    /// Shutdown all actors across all servers
    pub async fn shutdown_all_actors(&self) {
        info!(
            log_type = LogType::Supervisor.as_str(),
            "Shutting down all actors"
        );
        let mut actors = self.actors.write().await;

        for (config_id, pairs) in actors.iter_mut() {
            for pair in pairs.iter() {
                debug!(
                    log_type = LogType::Supervisor.as_str(),
                    "Shutting down actor for user {} on {}", pair.username, config_id
                );
                pair.shutdown
                    .store(true, std::sync::atomic::Ordering::SeqCst);
            }
        }

        // Give actors time to clean up
        tokio::time::sleep(tokio::time::Duration::from_millis(300)).await;

        // Clear all actors
        actors.clear();

        // Also clean up tasks
        let mut tasks = self.tasks.write().await;
        for (config_id, handles) in tasks.iter_mut() {
            debug!(
                log_type = LogType::Supervisor.as_str(),
                "Aborting {} tasks for {}",
                handles.len(),
                config_id
            );
            for handle in handles.iter() {
                handle.abort();
            }
        }
        tasks.clear();

        info!(
            log_type = LogType::Supervisor.as_str(),
            "All actors have been shut down"
        );
    }
}