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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
use std::{
cell::RefCell,
net::SocketAddr,
sync::{Arc, Weak},
};
use remote_control_rs::{client, tasks::Task};
use tokio::{
io::{self, AsyncReadExt},
net::{TcpListener, TcpStream},
};
use crate::servers::{
Server,
tcpserver::tasks::{Request, RequestType},
};
use bincode::serde;
struct TCPServer<'a> {
this: RefCell<Weak<Self>>,
task: Option<Task>,
listener: TcpListener,
clients: RefCell<Vec<Client<'a>>>,
}
impl<'a> TCPServer<'a> {
fn new(listener: TcpListener) -> Arc<Self> {
let s = Arc::new(Self { this: RefCell::new(Weak::new()), listener, task: None, clients: RefCell::new(Vec::new()) });
*s.this.borrow_mut() = Arc::downgrade(&s);
s
}
fn remove(&self, client: &Client<'a>) {
self.clients.borrow_mut().retain(|c| c != client);
}
}
impl<'a> Server<'a> for TCPServer<'a> {
fn update(&mut self, task: Task) {
self.task = Some(task);
}
async fn listen(&mut self) -> ! {
loop {
match self.listener.accept().await {
Ok((socket, addr)) => self.clients.borrow_mut().push(Client::new(self.this.borrow().upgrade().unwrap(), socket, addr)), // the unwrap shouldn't panics, self has to be alive
Err(e) => eprintln!("Failed to accept connection: {e}"),
};
}
}
}
struct Client<'a> {
server: Arc<TCPServer<'a>>,
stream: TcpStream,
addr: SocketAddr,
data: Option<client::Client<'a>>,
}
impl<'a> Client<'a> {
fn new(server: Arc<TCPServer<'a>>, stream: TcpStream, addr: SocketAddr) -> Self {
Self { server, stream, addr, data: None }
}
fn handle_read(&mut self, buf: io::Result<usize>) -> bool {
match buf {
Ok(0) => self.server.remove(self),
Ok(_) => {}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return false,
Err(e) => {
eprintln!("Failed to read from client: {e}");
return false;
}
}
true
}
// TODO: use framed
async fn read(&mut self) -> Option<(Vec<u8>, usize)> {
const usize: usize = std::mem::size_of::<usize>();
let mut len = vec![0; usize]; // vec! only take usize
if !self.handle_read(self.stream.try_read(&mut len)) {
return None;
}
let len = usize::from_be_bytes(match len.try_into() {
Ok(n) => n,
Err(e) => {
eprintln!("Failed to convert length bytes to length usize: {:?}", e);
return None;
}
}); // TODO: move to a shared fn with client to ensure we use same endian
let mut buf = vec![0; len];
let read = self.stream.read_exact(&mut buf).await;
if !self.handle_read(read) {
return None;
}
Some((buf, len))
}
async fn poll(&mut self) -> ! {
loop {
if let Err(e) = self.stream.readable().await {
eprintln!("Failed waiting for client to be readable: {e}");
continue;
}
if let Some((buf, len)) = self.read().await {
let req: RequestType = match serde::decode_from_slice(buf.as_slice(), bincode::config::standard()) {
Ok((decoded, size)) => {
if size != len {
eprintln!("Decoded size {size} does not match expected size {len}, decode succeeded anyways?");
}
decoded
}
Err(e) => {
eprintln!("Failed to decode request from client {}: {e}", self.addr);
continue;
}
};
match req {
// TODO: seperate send and respond
RequestType::Send(req) => match req {
Request::Get(req) => {
todo!()
}
Request::Post(req) => {
todo!()
}
},
RequestType::Respond(res) => match res {
Request::Get(req) => {
todo!()
}
Request::Post(req) => {
todo!()
}
},
}
}
}
}
}
impl<'a> PartialEq for Client<'a> {
fn eq(&self, other: &Self) -> bool {
self.data == other.data
}
}
|
