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
|
use crate::network::{
ConnectionInfo,
tcp::{TcpEvent, TcpEventCallback},
udp::PingRequest
};
use crate::state::{ExecutionContext, State};
use log::*;
use mumble_protocol::{Serverbound, control::ControlPacket};
use mumlib::command::{Command, CommandResponse};
use std::sync::{Arc, RwLock};
use tokio::sync::{mpsc, oneshot, watch};
pub async fn handle(
state: Arc<RwLock<State>>,
mut command_receiver: mpsc::UnboundedReceiver<(
Command,
oneshot::Sender<mumlib::error::Result<Option<CommandResponse>>>,
)>,
tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>,
ping_request_sender: mpsc::UnboundedSender<PingRequest>,
mut packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
mut connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
) {
debug!("Begin listening for commands");
while let Some((command, response_sender)) = command_receiver.recv().await {
debug!("Received command {:?}", command);
let mut state = state.write().unwrap();
let event = state.handle_command(command, &mut packet_sender, &mut connection_info_sender);
drop(state);
match event {
ExecutionContext::TcpEvent(event, generator) => {
let (tx, rx) = oneshot::channel();
//TODO handle this error
let _ = tcp_event_register_sender.send((
event,
Box::new(move |e| {
let response = generator(e);
response_sender.send(response).unwrap();
tx.send(()).unwrap();
}),
));
rx.await.unwrap();
}
ExecutionContext::Now(generator) => {
response_sender.send(generator()).unwrap();
}
ExecutionContext::Ping(generator, converter) => {
match generator() {
Ok(addr) => {
let res = ping_request_sender.send((
0,
addr,
Box::new(move |packet| {
response_sender.send(converter(packet)).unwrap();
}),
));
if res.is_err() {
panic!();
}
}
Err(e) => {
response_sender.send(Err(e)).unwrap();
}
};
}
}
}
}
|
