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
use crate::message;
use crate::stream;
use log::{error, info, warn};
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use tokio::sync::broadcast::Sender;
use uuid;
/// reads a `Msg` from the channel.
pub async fn read_channel_msg(
chan: Sender<message::Msg>,
) -> Result<message::Msg, tokio::sync::broadcast::error::RecvError> {
let mut rx = chan.subscribe();
rx.recv().await
}
/// Handels the communication between a client and the broker.
pub async fn handle_clinet(mut socket: TcpStream, chan: Sender<message::Msg>) {
let client_chan: tokio::sync::broadcast::Sender<message::Msg> =
tokio::sync::broadcast::Sender::new(1);
let client_id = uuid::Uuid::new_v4().to_string();
tokio::spawn(async move {
loop {
tokio::select! {
msg_ = stream::read_message(&mut socket) =>{
match msg_{
Ok(mut m)=>{
m.client_id(client_id.clone());
if !m.topic.is_empty() {
info!("topic: {}", m.topic);
match m.header.pkt_type{
message::PktType::PUBLISH=>{
info!("it's a publish packet");
match chan.send(m.clone()) {
Ok(n) => n,
Err(e) => {
error!("error while checking the topic in the map: {}",e.to_string());
0
}
};
},
message::PktType::SUBSCRIBE=>{
info!("it's a subscribe pkt attaching the channel");
m.channel(client_chan.clone());
match chan.send(m.clone()) {
Ok(n) => n,
Err(e) => {
error!("error while checking the topic in the map: {}",e.to_string());
0
}
};
}
message::PktType::UNSUBSCRIBE=> {
m.channel(client_chan.clone());
match chan.send(m.clone()) {
Ok(n) => n,
Err(e) => {
error!("error while checking the topic in the map: {}", e.to_string());
0
},
};
},
message::PktType::QUERY=>{
m.channel(client_chan.clone());
match chan.send(m.clone()) {
Ok(n) => n,
Err(e) => {
error!("error while checking the topic in the map: {}", e.to_string());
0
},
};
},
_=>{}
};
}
if m.header.pkt_type != message::PktType::QUERY{
match message::get_msg_response(m.clone()){
Ok(v)=>{
match socket.write_all(&v).await{
Ok(_)=>{},
Err(e)=>{
error!("could not write the data to the socket: {}", e.to_string());
}
}
},
Err(e)=>{
error!("error while writing the data to the socket: {}", e.to_string());
}
}
}
},
Err(_e)=>{
warn!("client disconnected: {}", client_id);
return;
}
};
},
chan_msg = read_channel_msg(client_chan.clone())=>{
match chan_msg {
Ok(m) => {
info!("message received: {:?}, {}", m.topic.clone(), m.message.len());
socket
.write_all(&m.bytes())
.await
.expect("failed to write data to socket");
},
Err(_e) => {},
}
}
}
}
});
}