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
|
mod audio;
mod command;
mod network;
mod notify;
mod state;
use crate::network::ConnectionInfo;
use crate::state::State;
use futures::join;
use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender};
use log::*;
use mumble_protocol::control::ControlPacket;
use mumble_protocol::crypt::ClientCryptState;
use mumble_protocol::voice::Serverbound;
use mumlib::command::{Command, CommandResponse};
use mumlib::setup_logger;
use std::fs;
use std::sync::{Arc, Mutex};
use tokio::sync::{mpsc, watch};
use tokio::task::spawn_blocking;
#[tokio::main]
async fn main() {
setup_logger(std::io::stderr(), true);
notify::init();
// check if another instance is live
let (tx_client, rx_client) =
ipc::channel::<mumlib::error::Result<Option<CommandResponse>>>().unwrap();
if let Ok(server_name) = fs::read_to_string(mumlib::SOCKET_PATH) {
if let Ok(tx0) = IpcSender::connect(server_name) {
if tx0.send((Command::Ping, tx_client)).is_ok() {
if matches!(rx_client.recv().unwrap(), Ok(Some(CommandResponse::Pong))) {
error!("Another instance of mumd is already running");
return;
} else {
warn!("Ping with no response. Continuing...");
}
}
}
}
// Oneshot channel for setting UDP CryptState from control task
// For simplicity we don't deal with re-syncing, real applications would have to.
let (crypt_state_sender, crypt_state_receiver) = mpsc::channel::<ClientCryptState>(1); // crypt state should always be consumed before sending a new one
let (packet_sender, packet_receiver) = mpsc::unbounded_channel::<ControlPacket<Serverbound>>();
let (command_sender, command_receiver) = mpsc::unbounded_channel::<(
Command,
IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
)>();
let (connection_info_sender, connection_info_receiver) =
watch::channel::<Option<ConnectionInfo>>(None);
let (response_sender, response_receiver) = mpsc::unbounded_channel();
let (ping_request_sender, ping_request_receiver) = mpsc::unbounded_channel();
let state = State::new(packet_sender, connection_info_sender);
let state = Arc::new(Mutex::new(state));
let (_, _, _, e, _) = join!(
network::tcp::handle(
Arc::clone(&state),
connection_info_receiver.clone(),
crypt_state_sender,
packet_receiver,
response_receiver,
),
network::udp::handle(
Arc::clone(&state),
connection_info_receiver.clone(),
crypt_state_receiver,
),
command::handle(
state,
command_receiver,
response_sender,
ping_request_sender,
),
spawn_blocking(move || {
// IpcSender is blocking
receive_oneshot_commands(command_sender);
}),
network::udp::handle_pings(ping_request_receiver),
);
e.unwrap();
}
fn receive_oneshot_commands(
command_sender: mpsc::UnboundedSender<(
Command,
IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
)>,
) {
loop {
// create listener
let (server, server_name): (
IpcOneShotServer<(
Command,
IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
)>,
String,
) = IpcOneShotServer::new().unwrap();
fs::write(mumlib::SOCKET_PATH, &server_name).unwrap();
debug!("Listening to {}", server_name);
// receive command and response channel
let (_, conn): (
_,
(
Command,
IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
),
) = server.accept().unwrap();
debug!("Sending command {:?} to command handler", conn.0);
command_sender.send(conn).unwrap();
}
}
|
