Start work on notifications from client, cargo fmt

src/notifications/events.rs

@@ -1,5 +1,5 @@
-use serde::{Deserialize, Serialize};
 use rauth::auth::Session;
+use serde::{Deserialize, Serialize};
 use snafu::Snafu;
 
 #[derive(Serialize, Deserialize, Debug, Snafu)]
@@ -15,7 +15,7 @@ pub enum WebSocketError {
 #[derive(Deserialize, Debug)]
 #[serde(tag = "type")]
 pub enum ServerboundNotification {
-    Authenticate(Session)
+    Authenticate(Session),
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -78,10 +78,9 @@ pub enum ClientboundNotification {
     GuildDelete {
         id: String,
     },*/
-
     UserRelationship {
         id: String,
         user: String,
         status: i32,
-    }
+    },
 }

src/notifications/hive.rs

@@ -1,12 +1,12 @@
 use super::events::ClientboundNotification;
 use crate::database::get_collection;
 
+use futures::FutureExt;
 use hive_pubsub::backend::mongo::MongodbPubSub;
 use hive_pubsub::PubSub;
+use log::{debug, error};
 use once_cell::sync::OnceCell;
 use serde_json::to_string;
-use futures::FutureExt;
-use log::{error, debug};
 
 static HIVE: OnceCell<MongodbPubSub<String, String, ClientboundNotification>> = OnceCell::new();
 
@@ -35,7 +35,7 @@ pub async fn listen() {
         .fuse()
         .await
         .expect("Hive hit an error");
-    
+
     dbg!("a");
 }
 

src/notifications/mod.rs

@@ -1,3 +1,3 @@
-pub mod websocket;
 pub mod events;
 pub mod hive;
+pub mod websocket;

src/notifications/websocket.rs

@@ -1,21 +1,26 @@
-use crate::util::variables::WS_HOST;
+use crate::{database::entities::User, util::variables::WS_HOST};
 
-use log::info;
-use ulid::Ulid;
-use async_std::task;
-use futures::prelude::*;
-use std::str::from_utf8;
-use std::sync::{Arc, RwLock};
-use many_to_many::ManyToMany;
-use std::collections::HashMap;
-use futures::stream::SplitSink;
-use async_tungstenite::WebSocketStream;
-use async_tungstenite::tungstenite::Message;
 use async_std::net::{TcpListener, TcpStream};
+use async_std::task;
+use async_tungstenite::tungstenite::Message;
+use futures::channel::mpsc::{unbounded, UnboundedSender};
+use futures::{pin_mut, prelude::*};
+use log::info;
+use many_to_many::ManyToMany;
+use rauth::auth::Session;
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::str::from_utf8;
+use std::sync::{Arc, Mutex, RwLock};
+use ulid::Ulid;
+
+use super::events::ServerboundNotification;
+
+type Tx = UnboundedSender<Message>;
+type PeerMap = Arc<Mutex<HashMap<SocketAddr, Tx>>>;
 
 lazy_static! {
-    static ref CONNECTIONS: Arc<RwLock<HashMap<String, SplitSink<WebSocketStream<TcpStream>, Message>>>> =
-        Arc::new(RwLock::new(HashMap::new()));
+    static ref CONNECTIONS: PeerMap = Arc::new(Mutex::new(HashMap::new()));
     static ref USERS: Arc<RwLock<ManyToMany<String, String>>> =
         Arc::new(RwLock::new(ManyToMany::new()));
 }
@@ -31,27 +36,41 @@ pub async fn launch_server() {
 }
 
 async fn accept(stream: TcpStream) {
-    let addr = stream.peer_addr().expect("Connected streams should have a peer address.");
+    let addr = stream
+        .peer_addr()
+        .expect("Connected streams should have a peer address.");
     let ws_stream = async_tungstenite::accept_async(stream)
         .await
         .expect("Error during websocket handshake.");
 
-    info!("User established WebSocket connection from {}.", addr);
+    info!("User established WebSocket connection from {}.", &addr);
 
     let id = Ulid::new().to_string();
     let (write, read) = ws_stream.split();
 
-    CONNECTIONS
-        .write()
-        .unwrap()
-        .insert(id, write);
+    let (tx, rx) = unbounded();
+    CONNECTIONS.lock().unwrap().insert(addr, tx);
 
-    read
-        .for_each(|message| async {
-            let data = message.unwrap().into_data();
-            // if you mess with the data, you get the bazooki
-            let string = from_utf8(&data).unwrap();
-            println!("{}", string);
-        })
-        .await;
+    let session: Option<Session> = None;
+    let user: Option<User> = None;
+
+    let fwd = rx.map(Ok).forward(write);
+    let reading = read.for_each(|message| async {
+        let data = message.unwrap().into_data();
+        // if you mess with the data, you get the bazooki
+        let string = from_utf8(&data).unwrap();
+
+        if let Ok(notification) = serde_json::from_str::<ServerboundNotification>(string) {
+            match notification {
+                ServerboundNotification::Authenticate(a) => {
+                    dbg!(a);
+                }
+            }
+        }
+    });
+
+    pin_mut!(fwd, reading);
+    future::select(fwd, reading).await;
+
+    println!("User {} disconnected.", &addr);
 }