+ Replacing rust microservice with more lenient js built server
The JSON-API can't _really_ use regular http requests because this server then has to do a lot of multi-threading nonsense. For the sake of simplicity for myself and others that try to write their own FC compliant servers: the rtc server(for now) only takes in websocket requests, and attemptes to discern servers from users connections for event handling
This commit is contained in:
shockrah
parent
afdeef0a49
commit
d8171f8b03
1
rtc-server/.gitignore
vendored
1
rtc-server/.gitignore
vendored
@ -1 +1,2 @@
|
||||
target/
|
||||
node_modules/
|
||||
|
||||
1126
rtc-server/Cargo.lock
generated
1126
rtc-server/Cargo.lock
generated
File diff suppressed because it is too large
Load Diff
@ -1,26 +0,0 @@
|
||||
[package]
|
||||
name = "rtc-server"
|
||||
version = "0.1.0"
|
||||
authors = ["shockrah <alejandros714@protonmail.com>"]
|
||||
edition = "2018"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
# Async infra stuff here
|
||||
futures = "0.3"
|
||||
tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread", "net", "macros", "fs"] }
|
||||
tokio-tungstenite = "0.14.0"
|
||||
|
||||
# Required for the private server interface
|
||||
hyper = { version = "0.14", features = ["full"] }
|
||||
|
||||
jsonwebtoken = "7"
|
||||
|
||||
# For caching mysql pools
|
||||
lazy_static = "1.4.0"
|
||||
|
||||
# For serialization of data in events
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
|
||||
clap = "2.33.3"
|
||||
@ -1,8 +0,0 @@
|
||||
default:
|
||||
cargo build --release
|
||||
|
||||
clean:
|
||||
cargo clean
|
||||
|
||||
run:
|
||||
cargo run --release
|
||||
50
rtc-server/auth.js
Normal file
50
rtc-server/auth.js
Normal file
@ -0,0 +1,50 @@
|
||||
const jsonwebtoken = require('jsonwebtoken')
|
||||
const fs = require('fs')
|
||||
|
||||
const SERVER_HMAC = fs.readFileSync('wss-hmac.secret')
|
||||
const USER_HMAC = fs.readFileSync('hmac.secret')
|
||||
|
||||
|
||||
exports.verify = function(token) {
|
||||
/**
|
||||
* @param {String} token
|
||||
* @returns 'user' on user connection
|
||||
* @returns 'server' on server connection
|
||||
* @retusn false on failure
|
||||
*/
|
||||
console.log('given token: ', token)
|
||||
try {
|
||||
const decoded = jsonwebtoken.verify(token, USER_HMAC, vconfig);
|
||||
return 'user'
|
||||
} catch (err) {
|
||||
try {
|
||||
const decoded = jsonwebtoken.verify(token, SERVER_HMAC, {ignoreNotBefore: true})
|
||||
return 'server'
|
||||
} catch (err) {
|
||||
console.log('failed server check: ', err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
exports.prepare_auth = function(req) {
|
||||
// NOTE: Why? because setting headers from the server is completely undocumented and I've ran
|
||||
// through basically every library under the sun I literally con't be fucked to
|
||||
// read people's code for a feature that could have a fucking tweet as documentation
|
||||
|
||||
// Typical User connections are setup with authentication in the headers
|
||||
// Requested channel is the path
|
||||
let header_auth = req.headers['authentication'] || req.headers['jwt']
|
||||
if(!header_auth) {
|
||||
let path = req.url
|
||||
let uri = '/jwt/'
|
||||
if(req.url.startsWith(uri)) {
|
||||
let jwt = req.url.slice(uri.length)
|
||||
return [jwt,null]
|
||||
}
|
||||
}
|
||||
else {
|
||||
return [header_auth, req.url]
|
||||
}
|
||||
}
|
||||
43
rtc-server/main.js
Normal file
43
rtc-server/main.js
Normal file
@ -0,0 +1,43 @@
|
||||
const ws = require('ws')
|
||||
const peersapi = require('./peers.js')
|
||||
const auth = require('./auth.js')
|
||||
|
||||
const server = new ws.Server({ port: 5648 })
|
||||
const peers = new peersapi.PeerMap()
|
||||
|
||||
|
||||
server.on('connection', function(ws, req) {
|
||||
let [jwt, path] = auth.prepare_auth(req)
|
||||
let conn = auth.verify(jwt)
|
||||
if(conn == 'server') {
|
||||
console.log('[WSS] New server connection')
|
||||
// Peer map initialization
|
||||
ws.on('open', function() {
|
||||
peers.add_server(ws)
|
||||
})
|
||||
ws.on('close', function() {
|
||||
console.log('server left')
|
||||
peers.remove_server(ws)
|
||||
})
|
||||
|
||||
ws.on('message', function(message) {
|
||||
peers.notify(message)
|
||||
})
|
||||
|
||||
} else if(conn == 'user') {
|
||||
ws.on('open', function() {
|
||||
console.log('adding with path', req.url)
|
||||
peers.add_user(ws, req.url)
|
||||
})
|
||||
|
||||
ws.on('close', function() {
|
||||
peers.remove_user(ws)
|
||||
})
|
||||
|
||||
console.log('[WSS] New user conncetion')
|
||||
} else {
|
||||
console.log('[WSS] No valid auth')
|
||||
ws.close()
|
||||
}
|
||||
|
||||
})
|
||||
112
rtc-server/package-lock.json
generated
Normal file
112
rtc-server/package-lock.json
generated
Normal file
@ -0,0 +1,112 @@
|
||||
{
|
||||
"name": "rtc-server",
|
||||
"version": "1.0.0",
|
||||
"lockfileVersion": 1,
|
||||
"requires": true,
|
||||
"dependencies": {
|
||||
"buffer-equal-constant-time": {
|
||||
"version": "1.0.1",
|
||||
"resolved": "https://registry.npmjs.org/buffer-equal-constant-time/-/buffer-equal-constant-time-1.0.1.tgz",
|
||||
"integrity": "sha1-+OcRMvf/5uAaXJaXpMbz5I1cyBk="
|
||||
},
|
||||
"ecdsa-sig-formatter": {
|
||||
"version": "1.0.11",
|
||||
"resolved": "https://registry.npmjs.org/ecdsa-sig-formatter/-/ecdsa-sig-formatter-1.0.11.tgz",
|
||||
"integrity": "sha512-nagl3RYrbNv6kQkeJIpt6NJZy8twLB/2vtz6yN9Z4vRKHN4/QZJIEbqohALSgwKdnksuY3k5Addp5lg8sVoVcQ==",
|
||||
"requires": {
|
||||
"safe-buffer": "^5.0.1"
|
||||
}
|
||||
},
|
||||
"jsonwebtoken": {
|
||||
"version": "8.5.1",
|
||||
"resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-8.5.1.tgz",
|
||||
"integrity": "sha512-XjwVfRS6jTMsqYs0EsuJ4LGxXV14zQybNd4L2r0UvbVnSF9Af8x7p5MzbJ90Ioz/9TI41/hTCvznF/loiSzn8w==",
|
||||
"requires": {
|
||||
"jws": "^3.2.2",
|
||||
"lodash.includes": "^4.3.0",
|
||||
"lodash.isboolean": "^3.0.3",
|
||||
"lodash.isinteger": "^4.0.4",
|
||||
"lodash.isnumber": "^3.0.3",
|
||||
"lodash.isplainobject": "^4.0.6",
|
||||
"lodash.isstring": "^4.0.1",
|
||||
"lodash.once": "^4.0.0",
|
||||
"ms": "^2.1.1",
|
||||
"semver": "^5.6.0"
|
||||
}
|
||||
},
|
||||
"jwa": {
|
||||
"version": "1.4.1",
|
||||
"resolved": "https://registry.npmjs.org/jwa/-/jwa-1.4.1.tgz",
|
||||
"integrity": "sha512-qiLX/xhEEFKUAJ6FiBMbes3w9ATzyk5W7Hvzpa/SLYdxNtng+gcurvrI7TbACjIXlsJyr05/S1oUhZrc63evQA==",
|
||||
"requires": {
|
||||
"buffer-equal-constant-time": "1.0.1",
|
||||
"ecdsa-sig-formatter": "1.0.11",
|
||||
"safe-buffer": "^5.0.1"
|
||||
}
|
||||
},
|
||||
"jws": {
|
||||
"version": "3.2.2",
|
||||
"resolved": "https://registry.npmjs.org/jws/-/jws-3.2.2.tgz",
|
||||
"integrity": "sha512-YHlZCB6lMTllWDtSPHz/ZXTsi8S00usEV6v1tjq8tOUZzw7DpSDWVXjXDre6ed1w/pd495ODpHZYSdkRTsa0HA==",
|
||||
"requires": {
|
||||
"jwa": "^1.4.1",
|
||||
"safe-buffer": "^5.0.1"
|
||||
}
|
||||
},
|
||||
"lodash.includes": {
|
||||
"version": "4.3.0",
|
||||
"resolved": "https://registry.npmjs.org/lodash.includes/-/lodash.includes-4.3.0.tgz",
|
||||
"integrity": "sha1-YLuYqHy5I8aMoeUTJUgzFISfVT8="
|
||||
},
|
||||
"lodash.isboolean": {
|
||||
"version": "3.0.3",
|
||||
"resolved": "https://registry.npmjs.org/lodash.isboolean/-/lodash.isboolean-3.0.3.tgz",
|
||||
"integrity": "sha1-bC4XHbKiV82WgC/UOwGyDV9YcPY="
|
||||
},
|
||||
"lodash.isinteger": {
|
||||
"version": "4.0.4",
|
||||
"resolved": "https://registry.npmjs.org/lodash.isinteger/-/lodash.isinteger-4.0.4.tgz",
|
||||
"integrity": "sha1-YZwK89A/iwTDH1iChAt3sRzWg0M="
|
||||
},
|
||||
"lodash.isnumber": {
|
||||
"version": "3.0.3",
|
||||
"resolved": "https://registry.npmjs.org/lodash.isnumber/-/lodash.isnumber-3.0.3.tgz",
|
||||
"integrity": "sha1-POdoEMWSjQM1IwGsKHMX8RwLH/w="
|
||||
},
|
||||
"lodash.isplainobject": {
|
||||
"version": "4.0.6",
|
||||
"resolved": "https://registry.npmjs.org/lodash.isplainobject/-/lodash.isplainobject-4.0.6.tgz",
|
||||
"integrity": "sha1-fFJqUtibRcRcxpC4gWO+BJf1UMs="
|
||||
},
|
||||
"lodash.isstring": {
|
||||
"version": "4.0.1",
|
||||
"resolved": "https://registry.npmjs.org/lodash.isstring/-/lodash.isstring-4.0.1.tgz",
|
||||
"integrity": "sha1-1SfftUVuynzJu5XV2ur4i6VKVFE="
|
||||
},
|
||||
"lodash.once": {
|
||||
"version": "4.1.1",
|
||||
"resolved": "https://registry.npmjs.org/lodash.once/-/lodash.once-4.1.1.tgz",
|
||||
"integrity": "sha1-DdOXEhPHxW34gJd9UEyI+0cal6w="
|
||||
},
|
||||
"ms": {
|
||||
"version": "2.1.3",
|
||||
"resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
|
||||
"integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="
|
||||
},
|
||||
"safe-buffer": {
|
||||
"version": "5.2.1",
|
||||
"resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
|
||||
"integrity": "sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ=="
|
||||
},
|
||||
"semver": {
|
||||
"version": "5.7.1",
|
||||
"resolved": "https://registry.npmjs.org/semver/-/semver-5.7.1.tgz",
|
||||
"integrity": "sha512-sauaDf/PZdVgrLTNYHRtpXa1iRiKcaebiKQ1BJdpQlWH2lCvexQdX55snPFyK7QzpudqbCI0qXFfOasHdyNDGQ=="
|
||||
},
|
||||
"ws": {
|
||||
"version": "7.4.4",
|
||||
"resolved": "https://registry.npmjs.org/ws/-/ws-7.4.4.tgz",
|
||||
"integrity": "sha512-Qm8k8ojNQIMx7S+Zp8u/uHOx7Qazv3Yv4q68MiWWWOJhiwG5W3x7iqmRtJo8xxrciZUY4vRxUTJCKuRnF28ZZw=="
|
||||
}
|
||||
}
|
||||
}
|
||||
24
rtc-server/package.json
Normal file
24
rtc-server/package.json
Normal file
@ -0,0 +1,24 @@
|
||||
{
|
||||
"name": "rtc-server",
|
||||
"version": "1.0.0",
|
||||
"description": "RTC for Freechat",
|
||||
"main": "main.js",
|
||||
"scripts": {
|
||||
"start": "node main.js",
|
||||
"test": "echo \"Error: no test specified\" && exit 1"
|
||||
},
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "git+https://gitlab.com/shockrah/freechat.git"
|
||||
},
|
||||
"author": "shockrah",
|
||||
"license": "GPL-3.0",
|
||||
"bugs": {
|
||||
"url": "https://gitlab.com/shockrah/freechat/issues"
|
||||
},
|
||||
"homepage": "https://gitlab.com/shockrah/freechat#readme",
|
||||
"dependencies": {
|
||||
"jsonwebtoken": "^8.5.1",
|
||||
"ws": "^7.4.4"
|
||||
}
|
||||
}
|
||||
69
rtc-server/peers.js
Normal file
69
rtc-server/peers.js
Normal file
@ -0,0 +1,69 @@
|
||||
class Event {
|
||||
/**
|
||||
* @param raw {String}
|
||||
*
|
||||
*/
|
||||
constructor(raw) {
|
||||
this._raw = raw
|
||||
let obj = JSON.parse(raw)
|
||||
this.type = obj['type']
|
||||
if(this.type == 'message') {
|
||||
this.message = obj['message']
|
||||
} else if (this.type == 'channel') {
|
||||
this.channel = obj['channel']
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class Peer {
|
||||
/*
|
||||
* @param type {String}
|
||||
* @param channel {String}
|
||||
*/
|
||||
constructor(type, channel) {
|
||||
this.type = type
|
||||
this.channel = channel
|
||||
}
|
||||
}
|
||||
|
||||
exports.PeerMap = class {
|
||||
/*
|
||||
* @field users Map<Socket, Peer>
|
||||
* @field server Map<Socket, Peer>
|
||||
*/
|
||||
constructor() {
|
||||
// Map of sockets -> Peer
|
||||
this.users = {}
|
||||
this.server = {}
|
||||
}
|
||||
|
||||
/**
|
||||
* @params message {String|JSON} Used for emitting server messages to user peers
|
||||
*/
|
||||
notify(message) {
|
||||
let e_type = new Event(message)
|
||||
if(e_type.type) {
|
||||
for(const conn of Object.keys(this.users)) {
|
||||
}
|
||||
} else {
|
||||
console.log('[WSS] Invalid event type given from server')
|
||||
}
|
||||
}
|
||||
|
||||
// As the names imply these methods help maintain the peer map
|
||||
add_server(socket) {
|
||||
this.server[socket] = new Peer('server', null)
|
||||
}
|
||||
remove_server(socket) {
|
||||
delete this.server[socket]
|
||||
}
|
||||
add_user(socket, channel) {
|
||||
this.users[socket] = new Peer('user', channel)
|
||||
}
|
||||
remove_user(socket) {
|
||||
delete this.user[socket]
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@ -1,87 +0,0 @@
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use serde::{Serialize, Deserialize, de::DeserializeOwned};
|
||||
use jsonwebtoken::{decode, DecodingKey};
|
||||
use jsonwebtoken::{Validation, Algorithm};
|
||||
use lazy_static::lazy_static;
|
||||
|
||||
use crate::peers::ConnectionType;
|
||||
|
||||
lazy_static! {
|
||||
// This hmac is the key we use to sign ONLY API <-> WSS communications
|
||||
static ref API_HMAC_SECRET: Vec<u8> = {
|
||||
std::fs::read("wss-hmac.secret").expect("[WSS FATAL] Couldn't get WSS HMAC secret")
|
||||
};
|
||||
|
||||
static ref USER_HMAC: Vec<u8> = {
|
||||
std::fs::read("hmac.secret").expect("[WSS FATAL] no user hmac.secret found")
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
trait Claims {
|
||||
/// Returns the unix timestamp in ms 0 if this field does not exist
|
||||
fn time(&self) -> i64;
|
||||
/// Returns user id of if one is present
|
||||
fn sub(&self) -> Option<u64>;
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize)]
|
||||
struct APIClaim;
|
||||
|
||||
impl Claims for APIClaim {
|
||||
fn time(&self) -> i64 { 0 }
|
||||
fn sub(&self) -> Option<u64> { None }
|
||||
}
|
||||
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct UserClaim {
|
||||
sub: u64, // user id
|
||||
exp: i64, // expiry date
|
||||
cookie: String, // unique cookie value
|
||||
}
|
||||
|
||||
impl Claims for UserClaim {
|
||||
fn time(&self) -> i64 { self.exp }
|
||||
fn sub(&self) -> Option<u64> { Some(self.sub) }
|
||||
}
|
||||
|
||||
|
||||
|
||||
fn verify_token<T>(token: &str) -> Option<ConnectionType> where
|
||||
T: DeserializeOwned + Claims
|
||||
{
|
||||
let dk = DecodingKey::from_secret(&USER_HMAC);
|
||||
let algo = Algorithm::HS512;
|
||||
|
||||
if let Ok(decoded) = decode::<T>(token, &dk, &Validation::new(algo)) {
|
||||
let time = decoded.claims.time();
|
||||
let now = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("[WSS UMMM] Couldn't get unix time")
|
||||
.as_millis() as i64;
|
||||
let active = now < time;
|
||||
match active {
|
||||
true => Some(ConnectionType::User),
|
||||
false => None
|
||||
}
|
||||
}
|
||||
// Only fallback to checking for a server connection since this basically never happens
|
||||
// compared to the user connection branch
|
||||
else {
|
||||
let dk = DecodingKey::from_secret(&API_HMAC_SECRET);
|
||||
if let Ok(_decoded) = decode::<T>(token, &dk, &Validation::new(algo)) {
|
||||
Some(ConnectionType::Server)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn verify(token: &str) -> Option<ConnectionType> {
|
||||
match verify_token::<UserClaim>(token) {
|
||||
Some(user) => Some(user),
|
||||
None => verify_token::<APIClaim>(token)
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,133 +0,0 @@
|
||||
mod auth;
|
||||
mod peers;
|
||||
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::collections::HashMap;
|
||||
|
||||
use futures::StreamExt;
|
||||
use futures::channel::mpsc::unbounded;
|
||||
use futures::future;
|
||||
use futures::pin_mut;
|
||||
use futures::stream::TryStreamExt;
|
||||
|
||||
use tokio_tungstenite as tokio_ws;
|
||||
use tokio_ws::tungstenite::handshake::server::ErrorResponse;
|
||||
use tokio_ws::tungstenite::http::{Response, Request};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
|
||||
use clap::{Arg, App};
|
||||
|
||||
use peers::{Peer, Channel, PeerMap, ConnectionType};
|
||||
|
||||
macro_rules! header_err {
|
||||
($s:literal) => {
|
||||
Err(ErrorResponse::new(Some($s.to_string())))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), std::io::Error> {
|
||||
let matches = App::new("Freechat RTC Websocket Server")
|
||||
.version("1.0")
|
||||
.author("shockrah")
|
||||
.about("Enables RTC communication to authorized client applications")
|
||||
.arg(Arg::with_name("port")
|
||||
.short("p")
|
||||
.long("port")
|
||||
.value_name("PORT")
|
||||
.takes_value(true))
|
||||
.get_matches();
|
||||
|
||||
let connections: PeerMap = Arc::new(Mutex::new(HashMap::new()));
|
||||
|
||||
// Websocket server initialization
|
||||
let wsaddr = match matches.value_of("port") {
|
||||
Some(pval) => {
|
||||
let port = pval.parse::<u16>().unwrap();
|
||||
format!("127.0.0.1:{}", port)
|
||||
},
|
||||
None => format!("127.0.0.1:5648")
|
||||
};
|
||||
let wssocket = TcpListener::bind(&wsaddr).await?;
|
||||
println!("[INFO] WSS Listening on {}", wsaddr);
|
||||
|
||||
while let Ok((stream, _)) = wssocket.accept().await {
|
||||
tokio::spawn(handle_connections(stream, connections.clone()));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
|
||||
async fn handle_connections(stream: TcpStream, peermap: PeerMap) {
|
||||
let addr = stream.peer_addr().expect("[ERROR] Peer address not found");
|
||||
|
||||
// NOTE: this call underneath actually blocks which blows but it will do for now
|
||||
// NOTE: find some kind of way of doing async callbacks here so that we can scale
|
||||
|
||||
let mut domain: Option<Channel> = None;
|
||||
let ws_stream = tokio_ws::accept_hdr_async(stream,
|
||||
|request:&Request<()>, response:Response<()>| -> Result<Response<()>, ErrorResponse> {
|
||||
|
||||
domain = match request.uri().path() {
|
||||
"/voice" => Some(Channel::Voice),
|
||||
"/text" => Some(Channel::Text),
|
||||
_ => None
|
||||
};
|
||||
|
||||
println!("[DEBUG] Incoming headers {:?}", request.headers());
|
||||
let entry = request.headers()
|
||||
.iter().find(|(name, _)| name.as_str() == "jwt");
|
||||
|
||||
if let Some((_, jwt)) = entry {
|
||||
match auth::verify(jwt.to_str().expect("Unable to convert header to str")) {
|
||||
Some(_conn_type) => Ok(response),
|
||||
None => panic!("[WSS] Unable to verify connection")
|
||||
}
|
||||
} else {
|
||||
header_err!("JWT not found in header")
|
||||
}
|
||||
}).await;
|
||||
|
||||
let ws_stream = match ws_stream {
|
||||
Ok(stream) => {
|
||||
println!("[WSS INFO] New connection established");
|
||||
stream
|
||||
}
|
||||
Err(e) => panic!(format!("[WSS ERROR] Could not finish handshake: {}", e))
|
||||
};
|
||||
|
||||
|
||||
let (tx, rx) = unbounded(); // split the peer's write and read streams
|
||||
let peer = Peer::new(tx, domain); // safe as the handshake fails with None domain
|
||||
println!("{:?}", peer);
|
||||
|
||||
// Add the new peer
|
||||
peermap.lock().unwrap().insert(addr, peer);
|
||||
|
||||
let (write, read) = ws_stream.split();
|
||||
|
||||
let broadcast_incoming = read.try_for_each(|msg| {
|
||||
let peers = peermap.lock().unwrap();
|
||||
// TODO: restructure this so that the server connection
|
||||
// never gets rans over to avoid this meme of a .collect
|
||||
// collect everyone except the server connection
|
||||
let recipients = peers
|
||||
.iter().filter(|(p_addr, meta)| p_addr != &&addr || meta.conn == ConnectionType::Server)
|
||||
.map(|(_, sink)| sink);
|
||||
|
||||
for recv in recipients {
|
||||
println!("{:?}", recv);
|
||||
recv.try_send_text(msg.clone());
|
||||
}
|
||||
future::ok(())
|
||||
});
|
||||
|
||||
// magic
|
||||
let forward = rx.map(Ok).forward(write);
|
||||
pin_mut!(broadcast_incoming, forward);
|
||||
future::select(broadcast_incoming, forward).await;
|
||||
println!("{} dc'd", &addr);
|
||||
peermap.lock().unwrap().remove(&addr);
|
||||
}
|
||||
@ -1,51 +0,0 @@
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::collections::HashMap;
|
||||
use std::net::SocketAddr;
|
||||
use futures::channel::mpsc::UnboundedSender;
|
||||
|
||||
use tokio_tungstenite as tokio_ws;
|
||||
use tokio_ws::tungstenite::Message;
|
||||
|
||||
|
||||
pub type PeerMap = Arc<Mutex<HashMap<SocketAddr, Peer>>>;
|
||||
type Tx = UnboundedSender<Message>;
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum ConnectionType {
|
||||
User,
|
||||
Server
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum Channel {
|
||||
Text,
|
||||
Voice,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Peer {
|
||||
transfer: Tx,
|
||||
channel: Option<Channel>,
|
||||
pub conn: ConnectionType
|
||||
}
|
||||
|
||||
impl Peer {
|
||||
pub fn new(transfer: Tx, channel: Option<Channel>) -> Self {
|
||||
Self {
|
||||
transfer,
|
||||
channel: channel.clone(),
|
||||
conn: match channel {
|
||||
Some(_) => ConnectionType::User,
|
||||
_ => ConnectionType::Server
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn try_send_text(&self, msg: Message) {
|
||||
if self.channel == Some(Channel::Text) {
|
||||
self.transfer.unbounded_send(msg)
|
||||
.expect("[WSS-COMM-ERROR] Unable to notify peer");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user