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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
use std::{
error::Error,
sync::{ Arc, atomic::{ self, AtomicUsize }},
net::SocketAddr,
convert::Infallible,
collections::HashMap,
};
mod minesweeper;
use minesweeper::*;
//use std::convert::{ TryFrom, TryInto };
use hyper::{ Method, StatusCode, Body, Request, Response, Server };
use hyper::service::{make_service_fn, service_fn};
use tokio::sync::{
RwLock,
mpsc,
};
type HtmlResult = Result<Response<Body>, Response<Body>>;
use futures_channel::mpsc::{unbounded, UnboundedSender};
use futures_util::{future, pin_mut, stream::TryStreamExt, StreamExt, sink::SinkExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::fs;
use tokio_tungstenite::tungstenite::protocol::Message;
type Tx = UnboundedSender<Message>;
type MovReqTx = mpsc::UnboundedSender<BoardRequest>;
type PeerMap = Arc<RwLock<HashMap<SocketAddr, (Tx, usize, String, (usize, usize))>>>;
type PeerInfo = (PeerMap, Arc::<AtomicUsize>);
// Which will either perform a move, or simply request the current board
type BoardRequest = Option<Move>;
const PAGE_RELPATH: &str = "./page.html";
const FONT_FILE_FUCKIT: &[u8] = include_bytes!("./VT323-Regular.ttf");
#[tokio::main]
async fn main() {
let sequential_id = Arc::new(AtomicUsize::new(0));
let peers = PeerMap::new(RwLock::new(HashMap::new()));
let peer_info: PeerInfo = (peers.clone(), sequential_id.clone());
let http_addr = SocketAddr::from(([0, 0, 0, 0], 31235));
println!("Http on {}", http_addr);
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
let game_t = tokio::spawn(gameloop(cmd_rx, peers.clone()));
// need to await this one at some point
let conn_l = tokio::spawn(conn_listener(peer_info.clone(), cmd_tx.clone()));
let http_serv = make_service_fn(|_| {
async move {
Ok::<_, Infallible>(service_fn(move |req: Request<Body>| {
async move {
Ok::<_,Infallible>(match handle_http_req(req).await {
Ok(r) => r,
Err(r) => r,
})
}
}))
}
});
let server = Server::bind(&http_addr)
.serve(http_serv)
.with_graceful_shutdown(shutdown_signal());
if let Err(e) = server.await {
eprint!("server error: {}", e);
}
}
// If a move is made, broadcast new board, else just send current board
async fn gameloop(mut move_rx: mpsc::UnboundedReceiver<BoardRequest>, peers: PeerMap) {
let mut game = Game::new(Board::new(30,10), (30*10)/8);
while let Some(req) = move_rx.recv().await {
let m = req;
if let Some(m) = m {
game = game.act(m);
}
let reply = Message::binary(game.board.render(None));
{
let peers = peers.read().await;
for (_, (tx, _, _, _)) in peers.iter() {
tx.unbounded_send(reply.clone()).unwrap();
}
}
}
}
async fn conn_listener(peer_info: PeerInfo, cmd_tx: MovReqTx) {
let ws_addr = SocketAddr::from(([0, 0, 0, 0], 31236));
let ws_socket = TcpListener::bind(&ws_addr).await;
let ws_listener = ws_socket.expect("Failed to bind");
// Let's spawn the handling of each connection in a separate task.
println!("Websocket on {}", ws_addr);
while let Ok((stream, addr)) = ws_listener.accept().await {
tokio::spawn(peer_connection(peer_info.clone(), cmd_tx.clone(), stream, addr));
}
}
async fn peer_connection(peer_info: PeerInfo, cmd_tx: MovReqTx, raw_stream: TcpStream, addr: SocketAddr) {
println!("Incoming TCP connection from: {}", addr);
let ws_stream = tokio_tungstenite::accept_async(raw_stream)
.await
.expect("Error during the websocket handshake occurred");
println!("WebSocket connection established: {}", addr);
let peer_map = peer_info.0;
let peer_seqid = peer_info.1.fetch_add(1, atomic::Ordering::AcqRel);
// Insert the write part of this peer to the peer map.
let (tx, rx) = unbounded();
peer_map.write().await.insert(addr, (tx.clone(), peer_seqid, "Dummy".to_string(), (0,0)));
let (outgoing, mut incoming) = ws_stream.split();
{ // 1 - Inform the new player its id
tx.unbounded_send(Message::text(format!("id {}", peer_seqid))).unwrap();
}
{ // 2 - Queue up a current board broadcast
cmd_tx.send(None).unwrap();
}
let process_incoming = async {
while let Some(cmd) = incoming.try_next().await.unwrap() {
let cmd = cmd.to_text().unwrap();
println!("Received a message from {}: {}", addr, cmd);
if cmd.starts_with("pos") {
let mut fields = cmd.split(" ").skip(1);
let pos = (fields.next().unwrap().parse::<usize>().unwrap(), fields.next().unwrap().parse::<usize>().unwrap());
let (name, id) = {
let mut peers = peer_map.write().await;
let mut entry = peers.get_mut(&addr).unwrap();
entry.3 = pos.clone();
(entry.2.clone(), entry.1)
};
{
let peers = peer_map.read().await;
for peer_tx in peers.iter().filter(|(s, _)| **s != addr).map(|(_,(peer_tx,_,_,_))| peer_tx) {
peer_tx.unbounded_send(Message::text(format!("pos {} {} {} {}", id, name, pos.0, pos.1))).unwrap();
}
}
}
}
};
let send_to_browser = rx.map(Ok).forward(outgoing);
pin_mut!(process_incoming, send_to_browser);
future::select(process_incoming, send_to_browser).await;
println!("{} disconnected", &addr);
peer_map.write().await.remove(&addr);
}
async fn handle_http_req(request: Request<Body>) -> HtmlResult {
let page = fs::read_to_string(PAGE_RELPATH).await.unwrap();
let mut uri_path = request.uri().path().split('/').skip(1);
let actual_path = uri_path.next();
match (request.method(), actual_path) {
(&Method::GET, None | Some("")) => {
Response::builder()
.status(StatusCode::OK)
.body(Body::from(page))
.map_err(errpage)
},
(&Method::GET, Some("saddr")) => {
Response::builder()
.status(StatusCode::OK)
.body(Body::from("placeholder"))
.map_err(errpage)
},
(&Method::GET, Some("font.ttf")) => {
Response::builder()
.status(StatusCode::OK)
.body(Body::from(FONT_FILE_FUCKIT))
.map_err(errpage)
},
_ => {
Response::builder()
.status(StatusCode::METHOD_NOT_ALLOWED)
.header("ALLOW", "GET, POST")
.body(Body::empty()).map_err(errpage)
}
}
}
fn errpage<T: Error>(e: T) -> Response<Body> {
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(e.to_string().into())
.unwrap()
}
async fn shutdown_signal() {
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C signal handler");
}
|
