Cargo.lock

@@ -609,19 +609,6 @@ dependencies = [
  "syn 2.0.15",
 ]
 
-[[package]]
-name = "dashmap"
-version = "5.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "907076dfda823b0b36d2a1bb5f90c96660a5bbcd7729e10727f07858f22c4edc"
-dependencies = [
- "cfg-if",
- "hashbrown",
- "lock_api",
- "once_cell",
- "parking_lot_core",
-]
-
 [[package]]
 name = "derive_more"
 version = "0.99.17"
@@ -1790,7 +1777,6 @@ dependencies = [
  "tauri-build",
  "tokio",
  "tonic",
- "uuid",
 ]
 
 [[package]]
@@ -1807,7 +1793,6 @@ name = "mraow-server"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "dashmap",
  "futures",
  "hyper",
  "mraow-common",
@@ -1820,7 +1805,6 @@ dependencies = [
  "tower",
  "tracing",
  "tracing-subscriber",
- "uuid",
 ]
 
 [[package]]

client/package.json

@@ -1,6 +1,5 @@
 {
   "scripts": {
-    "build": "vite build",
     "dev": "vite"
   },
   "devDependencies": {

client/src-tauri/Cargo.toml

@@ -21,7 +21,6 @@ mraow-common = { path = "../../common" }
 tokio = { version = "1.28.0", features = ["full"] }
 anyhow = "1.0.71"
 tonic = "0.9.2"
-uuid = { version = "1.3.2", features = ["v4"] }
 
 [features]
 default = ["mraow-common/client"]

client/src-tauri/src/main.rs

@@ -1,47 +1,21 @@
 // Prevents additional console window on Windows in release, DO NOT REMOVE!!
 #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
 
-use std::{sync::Arc, thread};
-
 use anyhow::Result;
-use mraow_common::chat_proto::{
-  chat_client::ChatClient, ChatMessage, ReceiveMsgsRequest, RoomAction,
-};
-use serde_json::json;
-use tauri::{
-  async_runtime::{Mutex, TokioHandle},
-  Manager, State,
-};
-use tonic::{transport::channel::Channel, IntoRequest};
-use uuid::Uuid;
+use mraow_common::chat_proto::chat_client::ChatClient;
+use tauri::async_runtime::{Mutex, TokioHandle};
+use tonic::transport::channel::Channel;
 
-type MyChatClient = Arc<Mutex<ChatClient<Channel>>>;
-
-pub struct UserId(Uuid);
+type MyGreeterClient = ChatClient<Channel>;
 
 #[tauri::command]
 async fn send_message(
-  state: State<'_, MyChatClient>,
-  user_id: State<'_, UserId>,
+  state: tauri::State<'_, Mutex<MyGreeterClient>>,
   message: String,
 ) -> Result<(), ()> {
   println!("SHIET {state:?}");
 
-  let mut client = state.lock().await;
-
-  let user_id = user_id.inner();
-  let resp = client
-    .send_msg(ChatMessage {
-      from_user_id: user_id.0.to_string(),
-      to_room_id: "general".to_string(),
-      content: message,
-      ..Default::default()
-    })
-    .await
-    .unwrap();
-
-  println!("Sent message to server. {resp:?}");
-
+  // let mut client = state.lock().await;
   /* client
   .say_hello(HelloRequest {
     message,
@@ -56,51 +30,15 @@ async fn send_message(
 async fn main() -> Result<()> {
   tauri::async_runtime::set(TokioHandle::current());
 
-  let uuid = Uuid::new_v4();
-  let user_id = UserId(uuid.clone());
-
-  let chat_client = ChatClient::connect("http://[::1]:50051").await?;
-  let chat_client = Arc::new(Mutex::new(chat_client));
-  let chat_client2 = chat_client.clone();
+  /*
+  TODO: Make sure this doesn't necessarily have to connect before starting the app
+  let greeter_client = ChatClient::connect("http://[::1]:50051").await?;
+  let greeter_client = Mutex::new(greeter_client);
+  println!("Connected :)");
+  */
 
   tauri::Builder::default()
-    .setup(move |app| {
-      let main_window = app.get_window("main").unwrap();
-
-      tokio::spawn(async move {
-        let mut client = chat_client2.lock().await;
-        client
-          .room_action(RoomAction {
-            room_id: "general".to_string(),
-            user_id: uuid.to_string(),
-            action: "join".to_string(),
-          })
-          .await
-          .unwrap();
-
-        let stream = client
-          .receive_msgs(ReceiveMsgsRequest {
-            user_id: uuid.to_string(),
-          })
-          .await
-          .unwrap();
-
-        std::mem::drop(client);
-
-        let mut stream = stream.into_inner();
-        while let Ok(Some(message)) = stream.message().await {
-          println!("SHIET message {message:?}");
-
-          main_window
-            .emit_all("new-message", json!({ "content" : message.content }))
-            .unwrap();
-        }
-      });
-
-      Ok(())
-    })
-    .manage(user_id)
-    .manage(chat_client)
+    // .manage(greeter_client)
     .invoke_handler(tauri::generate_handler![send_message])
     .run(tauri::generate_context!())
     .expect("error while running tauri application");

client/src-tauri/tauri.conf.json

@@ -29,7 +29,7 @@
         "icons/icon.icns",
         "icons/icon.ico"
       ],
-      "identifier": "io.mzhang.mraow",
+      "identifier": "com.tauri.dev",
       "longDescription": "",
       "macOS": {
         "entitlements": null,

client/src/App.tsx

@@ -1,5 +1,6 @@
 import { Provider } from "react-redux";
-import { store, useAppDispatch } from "./store";
+import { store } from "./store";
+import { useState } from "react";
 
 import styles from "./App.module.scss";
 import LeftSidebar from "./components/LeftSidebar";

client/src/components/CenterPanel.tsx

@@ -1,11 +1,9 @@
 import { invoke } from "@tauri-apps/api/tauri";
 import styles from "./CenterPanel.module.scss";
-import { useEffect, useState } from "react";
+import { useState } from "react";
 import { useAppDispatch, useAppSelector } from "../store";
 import { messageSelectors, messageSlice } from "../store/messages";
 import { v4 as uuidv4 } from "uuid";
-import { emit, listen } from "@tauri-apps/api/event";
-import { appWindow, WebviewWindow } from "@tauri-apps/api/window";
 
 export default function CenterPanel() {
   const [currentMessage, setCurrentMessage] = useState("");
@@ -14,32 +12,12 @@ export default function CenterPanel() {
     messageSelectors.selectAll(state)
   );
 
-  useEffect(() => {
-    let unlisten;
-
-    (async () => {
-      unlisten = await appWindow.listen("new-message", (event) => {
-        console.log("NEW EVENT", event);
-        const id = "lol";
-        const time = Date.now();
-        const content = event.payload.content;
-        dispatch(messageSlice.actions.addMessage({ id, time, content }));
-      });
-
-      console.log("Listen handler active.");
-    })();
-
-    return () => {
-      if (unlisten) unlisten();
-    };
-  });
-
   const onSubmit = (e) => {
     e.preventDefault();
     invoke("send_message", { message: currentMessage });
 
     const id = uuidv4();
-    const time = Date.now();
+    const time = new Date();
     dispatch(
       messageSlice.actions.addMessage({ id, time, content: currentMessage })
     );
@@ -53,8 +31,7 @@ export default function CenterPanel() {
       <div className={styles.middlePart}>
         {allMessages.map((msg) => (
           <div key={msg.id}>
-            <small>{new Date(msg.time).toISOString()}</small>
-            &nbsp;
+            <small>{msg.time.toISOString()}</small>
             {msg.content}
           </div>
         ))}

client/src/store/messages.ts

@@ -3,7 +3,7 @@ import { RootState } from ".";
 
 export type Message = {
   id: string;
-  time: number;
+  time: Date;
   content: string;
 };
 

client/tauri.conf.json5

@@ -0,0 +1,2 @@
+{
+}

client/vite.config.js

@@ -7,7 +7,7 @@ export default defineConfig({
 
   // Tauri expects a fixed port, fail if that port is not available
   server: {
-    strictPort: false,
+    strictPort: true,
   },
 
   // to make use of `TAURI_PLATFORM`, `TAURI_ARCH`, `TAURI_FAMILY`,

proto/chat.proto

@@ -12,21 +12,18 @@ message ChatMessage {
   string content = 5;
 }
 
-message Room { string name = 1; }
+message Room {
+  string name = 1;
+}
 
 message UserList { repeated User users = 1; }
 
 message RoomList {repeated Room rooms = 1;}
 
-message User {
-  string userId = 1;
-  string username = 2;
-}
+message User { string username = 1; }
 
 // RPC Messages
 
-message ReceiveMsgsRequest { string userId = 1; }
-
 message RoomActionResponse {}
 
 message RoomAction {
@@ -37,11 +34,11 @@ message RoomAction {
 
 service Chat {
   // Rooms
-  rpc roomAction(RoomAction) returns (google.protobuf.Empty) {}
+  rpc roomAction(RoomAction) returns (JoinResponse) {}
 
   // Messages
   rpc sendMsg(ChatMessage) returns (google.protobuf.Empty) {}
-  rpc receiveMsgs(ReceiveMsgsRequest) returns (stream ChatMessage) {}
+  rpc receiveMsgs(google.protobuf.Empty) returns (stream ChatMessage) {}
 
   // List
   rpc listRooms(google.protobuf.Empty) returns (RoomList) {}

server/Cargo.toml

@@ -20,8 +20,6 @@ tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 
 mraow-common = { path = "../common" }
-uuid = { version = "1.3.2", features = ["v4"] }
-dashmap = "5.4.0"
 
 [build-dependencies]
 tonic-build = "0.9.2"

server/src/main.rs

@@ -2,12 +2,6 @@ mod mux;
 
 use futures::Stream;
 use hyper::Body;
-use mraow_common::chat_proto::{
-  chat_server::{Chat, ChatServer},
-  ChatMessage, ReceiveMsgsRequest, RoomAction, RoomActionResponse, RoomList,
-  UserList,
-};
-use mux::{SharedMux, TextMessage};
 use std::{
   pin::Pin,
   task::{Context, Poll},
@@ -15,17 +9,17 @@ use std::{
 };
 use tonic::{body::BoxBody, transport::Server, Request, Response, Status};
 use tower::{Layer, Service};
-use uuid::Uuid;
 
-use crate::mux::{RoomId, RoomMessage, UserId};
+use mraow_common::chat_proto::{
+  chat_server::{Chat, ChatServer},
+  ChatMessage, RoomAction, RoomActionResponse, RoomList, User, UserList,
+};
 
-pub type ResponseStream =
+type ResponseStream =
   Pin<Box<dyn Stream<Item = Result<ChatMessage, Status>> + Send>>;
 
 #[derive(Default)]
-pub struct ChatImpl {
-  mux: SharedMux,
-}
+pub struct ChatImpl {}
 
 #[tonic::async_trait]
 impl Chat for ChatImpl {
@@ -34,49 +28,25 @@ impl Chat for ChatImpl {
   async fn room_action(
     &self,
     request: Request<RoomAction>,
-  ) -> Result<Response<()>, Status> {
-    let request = request.into_inner();
+  ) -> Result<Response<RoomActionResponse>, Status> {
     println!("Join {request:?}");
-
-    let user_id = Uuid::parse_str(&request.user_id).unwrap();
-    let user_id = UserId(user_id);
-    let room_id = RoomId(request.room_id);
-    self.mux.user_join_room(user_id, room_id);
-
-    Ok(Response::new(()))
+    todo!()
   }
 
   async fn send_msg(
     &self,
     request: Request<ChatMessage>,
   ) -> Result<Response<()>, Status> {
-    let request = request.into_inner();
     println!("Send message {request:?}");
-
-    // TODO: Get room id
-    let room_id = RoomId(request.to_room_id);
-    let message = RoomMessage::Text(TextMessage {
-      content: request.content,
-    });
-
-    self.mux.send_message_to_room(room_id, message);
-
-    Ok(Response::new(()))
+    todo!()
   }
 
-  async fn receive_msgs(
+  async fn receive_msg(
     &self,
-    request: Request<ReceiveMsgsRequest>,
-  ) -> Result<Response<Self::receiveMsgsStream>, Status> {
-    let request = request.into_inner();
+    request: Request<()>,
+  ) -> Result<Response<Self::receiveMsgStream>, Status> {
     println!("Receive message {request:?}");
-
-    let user_id = Uuid::parse_str(&request.user_id).unwrap();
-    let user_id = UserId(user_id);
-    let stream = self.mux.create_listen_stream(user_id);
-
-    // TODO: Handle drop
-    Ok(Response::new(stream))
+    todo!()
   }
 
   async fn list_rooms(
@@ -94,6 +64,27 @@ impl Chat for ChatImpl {
     println!("Get all users {request:?}");
     todo!()
   }
+
+  /*
+  async fn say_hello(
+    &self,
+    request: Request<HelloRequest>,
+  ) -> Result<Response<HelloReply>, Status> {
+    let addr = request.remote_addr();
+    let request = request.into_inner();
+
+    println!(
+      "Got a request from {:?}: {}",
+      addr,
+      request.message
+    );
+
+    let reply = HelloReply {
+      message: format!("Hello {}!", request.name),
+    };
+    Ok(Response::new(reply))
+  }
+  */
 }
 
 #[tokio::main]

server/src/mux.rs

@@ -1,137 +1,106 @@
 use std::{
-  ops::{Deref, DerefMut},
+  collections::HashMap,
   pin::Pin,
-  sync::Arc,
+  task::{Context, Poll},
 };
 
-use dashmap::{DashMap, DashSet};
-
-use futures::{stream, FutureExt, Stream, StreamExt};
-use mraow_common::chat_proto::{
-  chat_server::{Chat, ChatServer},
-  ChatMessage, ReceiveMsgsRequest, RoomAction, RoomActionResponse, RoomList,
-  UserList,
-};
-use tokio::sync::{
+use anyhow::Result;
+use futures::{future, Future, FutureExt, Sink, Stream};
+use tokio::{
+  sync::{
+    broadcast::{Receiver, Sender},
     mpsc::{self, UnboundedReceiver, UnboundedSender},
-  Mutex,
+  },
+  task::JoinHandle,
 };
-use tokio_stream::wrappers::UnboundedReceiverStream;
-use uuid::Uuid;
 
-use crate::ResponseStream;
-
-#[derive(Debug)]
-pub struct TextMessage {
-  pub content: String,
-}
-
-#[derive(Debug)]
-pub enum RoomMessage {
-  Text(TextMessage),
-}
-
-#[derive(Debug, Clone, Hash, PartialEq, Eq)]
-pub struct UserId(pub Uuid);
-
-#[derive(Debug, Clone, Hash, PartialEq, Eq)]
-pub struct RoomId(pub String);
-
-#[derive(Clone, Default)]
-pub struct SharedMux(Arc<Mux>);
-
-impl Deref for SharedMux {
-  type Target = Mux;
-
-  fn deref(&self) -> &Self::Target {
-    &self.0
-  }
-}
+#[derive(Clone)]
+pub struct SharedMux(Mutex<Mux>);
 
 /// Shared server state
-#[derive(Default)]
 pub struct Mux {
-  rooms: DashMap<RoomId, Room>,
-  users: DashMap<UserId, ConnectedUser>,
+  rooms: HashMap<String, Room>,
 }
 
-impl Mux {
-  pub fn user_join_room(&self, user_id: UserId, room_id: RoomId) {
-    self
-      .rooms
-      .entry(room_id)
-      .and_modify(|room| {
-        room.members.insert(user_id.clone());
-      })
-      .or_insert_with(|| {
-        let members = DashSet::new();
-        members.insert(user_id);
-        Room { members }
-      });
+#[derive(Clone)]
+pub enum RoomMessage {}
 
-    println!("Joined room. Rooms: {:?}", self.rooms);
-  }
-
-  pub fn send_message_to_room(&self, room_id: RoomId, message: RoomMessage) {
-    let message = Arc::new(message);
-
-    let recipients =
-      match self.rooms.get(&room_id).map(|room| room.members.clone()) {
-        Some(v) => v,
-        None => return,
-      };
-
-    println!("Sending message {message:?} to recipients:");
-    for user_id in recipients {
-      // TODO: This should technically never be None?
-      let user = self.users.get(&user_id).unwrap();
-      println!(" - user: {user:?}");
-
-      user.tx.send(message.clone());
-    }
-  }
-
-  pub fn ensure_user(&self, user_id: UserId) {
-    self.users.entry(user_id).or_insert_with(|| {
-      let (tx, rx) = mpsc::unbounded_channel();
-      ConnectedUser { tx, rx: Some(rx) }
-    });
-  }
-
-  pub fn create_listen_stream(&self, user_id: UserId) -> ResponseStream {
-    let rx = {
-      self.ensure_user(user_id.clone());
-      let mut user = self.users.get_mut(&user_id).unwrap();
-      user.rx.take().unwrap()
-    };
-    let stream = UnboundedReceiverStream::new(rx);
-
-    stream
-      .map(move |message| {
-        println!("Sending message to {user_id:?}: {message:?}");
-
-        let message = match message.as_ref() {
-          RoomMessage::Text(v) => v,
-        };
-
-        let chat_message = ChatMessage {
-          content: message.content.clone(),
-          ..Default::default()
-        };
-
-        Ok(chat_message)
-      })
-      .boxed()
-  }
-}
-
-#[derive(Debug)]
 pub struct Room {
-  members: DashSet<UserId>,
+  name: String,
+  tx: Sender<RoomMessage>,
+  rx: Receiver<RoomMessage>,
 }
 
-#[derive(Debug)]
-pub struct ConnectedUser {
-  tx: UnboundedSender<Arc<RoomMessage>>,
-  rx: Option<UnboundedReceiver<Arc<RoomMessage>>>,
+impl Room {
+  pub fn get_handle(&self) -> RoomHandle {
+    RoomHandle {
+      name: self.name.clone(),
+      tx: self.tx.clone(),
+      rx: self.tx.subscribe(),
+    }
+  }
+}
+
+pub struct RoomHandle {
+  name: String,
+  tx: Sender<RoomMessage>,
+  rx: Receiver<RoomMessage>,
+}
+
+pub enum CompositeListenerMessage {
+  Room { name: String, message: RoomMessage },
+  JoinRoom(String),
+  DropRoom(String),
+}
+
+pub struct CompositeListener {
+  current_listening_rooms: HashMap<String, RoomHandle>,
+}
+
+impl CompositeListener {}
+
+/// Create a "composite" listener that listens to multiple rooms / streams, and
+/// is able to add more streams in the middle as well
+pub fn create_listener() -> (
+  JoinHandle<()>,
+  UnboundedSender<CompositeListenerMessage>,
+  UnboundedReceiver<CompositeListenerMessage>,
+) {
+  let (tx, rx) = mpsc::unbounded_channel();
+
+  /*
+  let join_handle = tokio::spawn(async {
+    let mut currently_listening_to: HashMap<String, RoomHandle> =
+      HashMap::new();
+
+    loop {
+      // Build select targets
+      let ouais = rx.recv().boxed();
+      let mut select_targets: Vec<
+        Pin<Box<dyn Future<Output = Option<CompositeListenerMessage>>>>,
+      > = vec![ouais];
+      for (_, room_handle) in currently_listening_to.iter() {
+        select_targets.push(
+          room_handle
+            .rx
+            .recv()
+            .map(|f| {
+              f.ok().map(|m| CompositeListenerMessage::Room {
+                name: room_handle.name.clone(),
+                message: m,
+              })
+            })
+            .boxed(),
+        );
+      }
+
+      // Select
+      let result = future::select_all(select_targets).await;
+    }
+  });
+
+  (join_handle, tx, rx)
+  */
+
+  todo!()
 }