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
|
use std::time::Duration;
use crate::protocol::ClientboundPacket;
use crate::{crypto::hash, protocol::ServerboundPacket};
use futures_util::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use tokio_tungstenite::tungstenite::{self, Message};
pub async fn signaling_connect(
host: &str,
secret: &str,
) -> (
UnboundedSender<ServerboundPacket>,
UnboundedReceiver<ClientboundPacket>,
) {
let uri = format!("wss://{host}/signaling/{}", hash(secret));
info!("connecting to signaling server at {uri:?}");
let (conn, _) = tokio_tungstenite::connect_async(url::Url::parse(&uri).unwrap())
.await
.unwrap();
info!("connection established");
let (mut tx, mut rx) = conn.split();
let (in_tx, in_rx) = unbounded_channel();
let (out_tx, mut out_rx) = unbounded_channel();
let ping_out_tx = out_tx.clone();
let ping_task = tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
ping_out_tx.send(ServerboundPacket::Ping).unwrap()
}
});
let send_task = tokio::spawn(async move {
while let Some(p) = out_rx.recv().await {
debug!(" -> {p:?}");
tx.send(Message::Text(
serde_json::to_string::<ServerboundPacket>(&p).unwrap(),
))
.await
.unwrap();
}
});
let _receive_task = tokio::spawn(async move {
while let Some(mesg) = rx.next().await {
match mesg {
Ok(mesg) => match mesg {
tungstenite::Message::Text(t) => {
let p: ClientboundPacket = serde_json::from_str(t.as_str()).unwrap();
debug!("<- {p:?}");
in_tx.send(p).unwrap()
}
tungstenite::Message::Close(e) => {
error!("ws closed :( {e:?}");
unreachable!();
}
_ => (),
},
Err(_) => {
send_task.abort();
ping_task.abort();
break;
}
}
}
warn!("recv task stopped");
});
(out_tx, in_rx)
}
|
