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
|
//! A TCP-connection from a client to the server.
use super::super::packet::{Packet, PacketRwError};
use super::connection_manager::ConnId;
use serde::de::DeserializeOwned;
use serde::Serialize;
use std::fmt::Debug;
use std::net::TcpStream;
use std::sync::mpsc::Sender;
use std::thread::{self, JoinHandle};
/// Holds a stream to the client and manages receiving packets from said client
/// in an extra thread.
pub struct Connection {
stream: TcpStream,
rcv_thread_handle: JoinHandle<()>,
}
impl Connection {
/// Start up the receiving thread after a client connection has been detected.
/// Will create a ghost disconnect packet in case the client disconnects.
pub(super) fn start_rcv<P>(
conn_id: ConnId,
stream: TcpStream,
packet_tx: Sender<(ConnId, Packet<P>)>,
) -> Self
where
P: 'static + Debug + Send + DeserializeOwned + Serialize,
{
let mut stream_cl = stream
.try_clone()
.expect("Unable to clone TcpStream handle");
let rcv_thread_handle = thread::spawn(move || {
let mut running = true;
while running {
// Read the newest packet from the stream.
let packet = match Packet::read_from_stream(&mut stream_cl) {
Ok(packet) => dbg!(packet),
Err(PacketRwError::Closed) => {
// Stop the thread after this packet.
running = false;
/* Generate an internal disconnection packet, so the connection
* manager can call cleanup code if necessary.
*/
Packet::Disconnect
}
Err(err) => {
error!(
"Receiving packet failed. Connection `{}`. {:?}",
conn_id, err
);
// Ignore the received data.
continue;
}
};
/* Try sending the packet to the Connection manager. If it has already
* stopped and hung up on the channel, stop this receive thread as well.
*/
if packet_tx.send((conn_id, packet)).is_err() {
info!("Shutting down connection `{}`", conn_id);
running = false;
}
}
info!("Packet receive thread has stopped running.");
});
Self {
stream,
rcv_thread_handle,
}
}
/// Send a packet to the client via TCP.
pub(super) fn send<P>(&mut self, packet: &Packet<P>) -> Result<(), PacketRwError>
where
P: 'static + Debug + Send + DeserializeOwned + Serialize,
{
packet.write_to_stream(&mut self.stream)
}
}
|
