demostf/sync-rs
1
0
Fork 0
mirror of https://codeberg.org/demostf/sync.git synced 2026-06-03 16:44:07 +02:00
Code Issues Releases Wiki Activity

use a more maintained websocket library and cleanup on socket close

Browse source
This commit is contained in:
Robin Appelman 2017-10-07 17:24:44 +02:00
commit 2f59220df5
3 changed files with 225 additions and 149 deletions
Download patch file Download diff file Expand all files Collapse all files

112
src/main.rs
View file

@ -1,18 +1,16 @@
#[macro_use]
extern crate serde_derive;
extern crate mio_websocket;
extern crate ws;
extern crate mio;
extern crate serde;
extern crate serde_json;
use std::net::SocketAddr;
use mio::Token;
use mio_websocket::interface::*;
use std::collections::HashMap;
use std::collections::HashSet;
use std::borrow::Borrow;
use std::iter::FromIterator;
use ws::{listen, Handler, Sender, Result, Message, CloseCode, Error};
use std::rc::Rc;
use std::cell::RefCell;
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "type")]
@ -29,17 +27,18 @@ enum SyncCommand {
struct Session {
owner: Token,
clients: HashSet<Token>,
clients: HashMap<Token, Sender>,
tick: u64,
playing: bool
}
fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>, client: Token, ws: &mut WebSocket) {
fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>, client: Sender) {
match command {
SyncCommand::CreateCommand { session } => {
let new_session = Session {
owner: client,
clients: HashSet::new(),
owner: client.token(),
clients: HashMap::new(),
playing: false,
tick: 0
};
@ -48,15 +47,15 @@ fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>,
SyncCommand::JoinCommand { session: session_name } => {
match sessions.get_mut(&session_name) {
Some(mut session) => {
session.clients.insert(client);
session.clients.insert(client.token(), client);
send_to_session(session, &SyncCommand::TickPacket {
tick: session.tick,
session: session_name.clone()
}, ws);
});
send_to_session(session, &SyncCommand::PlayPacket {
play: session.playing,
session: session_name.clone()
}, ws);
});
}
None => println!("session {} not found", session_name)
}
@ -64,12 +63,12 @@ fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>,
SyncCommand::TickPacket { tick, session: session_name } => {
match sessions.get_mut(&session_name) {
Some(mut session) => {
if session.owner == client {
if session.owner == client.token() {
session.tick = tick;
send_to_session(session, &SyncCommand::TickPacket {
tick,
session: session_name
}, ws);
});
}
}
None => println!("session {} not found", session_name)
@ -78,12 +77,12 @@ fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>,
SyncCommand::PlayPacket { play, session: session_name } => {
match sessions.get_mut(&session_name) {
Some(mut session) => {
if session.owner == client {
if session.owner == client.token() {
session.playing = play;
send_to_session(session, &SyncCommand::PlayPacket {
play,
session: session_name
}, ws);
});
}
}
None => println!("session {} not found", session_name)
@ -92,47 +91,58 @@ fn handle_command(command: SyncCommand, sessions: &mut HashMap<String, Session>,
}
}
fn send_to_session(session: &Session, command: &SyncCommand, ws: &mut WebSocket) {
fn send_to_session(session: &Session, command: &SyncCommand) {
let command_text = serde_json::to_string(command).unwrap();
let all_clients = HashSet::from_iter(ws.get_connected().unwrap());
let peers = &all_clients & &session.clients;
for peer in peers {
let response = WebSocketEvent::TextMessage(command_text.clone());
ws.send((peer, response));
for client in session.clients.values() {
client.send(Message::from(command_text.clone())).ok();
}
}
struct Server {
out: Sender,
sessions: Rc<RefCell<HashMap<String, Session>>>,
}
impl Handler for Server {
fn on_message(&mut self, msg: Message) -> Result<()> {
let result: serde_json::Result<SyncCommand> = serde_json::from_str(msg.as_text().unwrap_or_default());
match result {
Ok(command) => {
handle_command(command, &mut self.sessions.borrow_mut(), self.out.clone());
Ok(())
}
Err(_) => Ok(())
}
}
fn on_close(&mut self, _: CloseCode, _: &str) {
let mut sessions = self.sessions.borrow_mut();
let token = self.out.token();
let owned_sessions: Vec<_> = sessions
.iter()
.filter(|&(_, v)| v.owner == token)
.map(|(k, _)| k.clone())
.collect();
for empty in owned_sessions { sessions.remove(&empty); }
for session in sessions.values_mut() {
session.clients.remove(&token);
}
}
fn on_error(&mut self, err: Error) {
println!("The server encountered an error: {:?}", err);
}
}
fn main() {
let port = std::env::var("PORT").unwrap_or("80".to_string());
let listen = format!("0.0.0.0:{}", port);
let mut sessions: HashMap<String, Session> = HashMap::new();
let listen_adress = format!("0.0.0.0:{}", port);
let mut ws = WebSocket::new(listen.parse::<SocketAddr>().unwrap());
println!("listening on: {:?}", listen_adress);
println!("listening on: {:?}", listen);
let sessions: Rc<RefCell<HashMap<String, Session>>> = Rc::new(RefCell::new(HashMap::new()));
loop {
match ws.next() {
(tok, WebSocketEvent::Connect) => {
println!("connected peer: {:?}", tok);
}
(tok, WebSocketEvent::TextMessage(msg)) => {
let result: Result<SyncCommand, serde_json::Error> = serde_json::from_str(msg.borrow());
match result {
Ok(command) => {
handle_command(command, &mut sessions, tok, &mut ws);
}
_ => {}
}
}
(tok, WebSocketEvent::BinaryMessage(msg)) => {
let response = WebSocketEvent::BinaryMessage(msg.clone());
ws.send((tok, response));
}
_ => {}
}
}
listen(listen_adress, |out| { Server { out, sessions: sessions.clone() } }).unwrap()
}