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
//! Serving one panel: events and replies out, commands in.
use futures_util::{SinkExt as _, StreamExt as _};
use rs_teststand_bridge::{Ack, Command, Response};
use std::sync::mpsc;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use super::{Outbound, Request};
/// Serves one panel: events and replies out, commands in, until it leaves.
pub(super) async fn serve_client<S>(
websocket: tokio_tungstenite::WebSocketStream<S>,
client: u64,
mut outbound: broadcast::Receiver<Outbound>,
commands: mpsc::Sender<Request>,
) where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let (mut sink, mut stream) = websocket.split();
loop {
tokio::select! {
// Something to send this panel.
received = outbound.recv() => {
let Ok(message) = received else {
// Lagged past the backlog, or the host is finished. Either
// way this panel is done: silently missing messages is worse
// than a close it can react to.
break;
};
if matches!(message, Outbound::Shutdown) {
// Leaving the loop reaches the `sink.close()` below, which
// sends the Close the peer is owed.
break;
}
let text = match message {
Outbound::Event(event) => serde_json::to_string(&event),
// A reply belongs to one panel; the others skip it.
Outbound::Reply { client: target, response } if target == client => {
// Converted at the wire boundary, so a host keeps
// building `Response` while every client receives the
// fixed five-field acknowledgement.
serde_json::to_string(&Ack::from(response.as_ref()))
}
Outbound::Reply { .. } => continue,
// Handled above; the match needs the arm to stay total.
Outbound::Shutdown => break,
};
let Ok(text) = text else { continue };
if sink.send(Message::text(text)).await.is_err() {
break;
}
}
// Something the panel asked for.
received = stream.next() => {
let Some(Ok(frame)) = received else { break };
match frame {
Message::Text(text) => match serde_json::from_str::<Command>(&text) {
Ok(command) => {
if commands.send(Request { client, command }).is_err() {
// The host is gone; nothing left to serve.
break;
}
}
// A malformed command is the panel's mistake, not a
// reason to drop it: say so and carry on.
Err(error) => {
let reply = Response::Failed {
command: "unparsed".to_owned(),
reason: error.to_string(),
};
let Ok(text) = serde_json::to_string(&reply) else { continue };
if sink.send(Message::text(text)).await.is_err() {
break;
}
}
},
// Breaking here reaches the `sink.close()` below, which
// sends the Close that RFC 6455 section 5.5.1 requires in
// reply. Returning early instead would leave the peer
// waiting for it.
Message::Close(_) => break,
// Ping and pong are answered by the library; binary frames
// are not part of this protocol.
_ => {}
}
}
}
}
// Completes the closing handshake, whether this side started it or the
// client did. A failure means the peer has already gone.
let _ = sink.close().await;
}