Implement basic connection establishing

src/connection.rs

@@ -0,0 +1,93 @@
+use std::sync::Arc;
+
+use tokio::io::{AsyncRead, AsyncWrite};
+
+use crate::db::{Db, User};
+use crate::proto::mumble::{ChannelState, Ping, UserState, Version};
+use crate::protocol::{MUMBLE_PROTOCOL_VERSION, MumblePacket, MumblePacketStream};
+
+pub struct Connection<S> {
+    db: Arc<Db>,
+    stream: MumblePacketStream<S>,
+}
+
+pub enum Error {
+    ConnectionSetupError,
+    AuthenticationError,
+}
+
+impl<S> Connection<S>
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send,
+{
+    pub async fn setup_connection(db: Arc<Db>, mut stream: MumblePacketStream<S>) -> Result<Connection<S>, Error> {
+        let _ = match stream.read().await? {
+            MumblePacket::Version(version) => version,
+            _ => return Err(Error::ConnectionSetupError)
+        };
+        let mut version = Version::new();
+        version.set_version(MUMBLE_PROTOCOL_VERSION);
+        stream.write(MumblePacket::Version(version)).await?;
+
+        let auth = match stream.read().await? {
+            MumblePacket::Authenticate(auth) => auth,
+            _ => return Err(Error::ConnectionSetupError)
+        };
+        if !auth.has_username() {
+            return Err(Error::AuthenticationError);
+        }
+        db.add_new_user(User {
+            username: auth.get_username().to_string(),
+            channel_id: 0,
+            is_connected: true,
+        }).await;
+
+        //TODO UDP crypt setup
+
+        let channels = db.get_channels();
+        for channel in channels {
+            let mut channel_state = ChannelState::new();
+            channel_state.set_name(channel.name);
+            channel_state.set_channel_id(channel.id);
+            stream.write(MumblePacket::ChannelState(channel_state)).await?;
+        }
+
+        let connected_users = db.get_connected_users();
+        for user in connected_users {
+            let mut user_state = UserState::new();
+            user_state.set_name(user.username);
+            user_state.set_channel_id(user.channel_id);
+            stream.write(MumblePacket::UserState(user_state)).await?;
+        }
+
+        Ok(Connection {
+            db,
+            stream,
+        })
+    }
+
+    pub async fn read_packet(&mut self) -> Result<MumblePacket, Error> {
+        Ok(self.stream.read().await?)
+    }
+
+    pub async fn handle_packet(&mut self, packet: MumblePacket) -> Result<(), Error> {
+        match packet {
+            MumblePacket::Ping(ping) => {
+                if ping.has_timestamp() {
+                    let mut ping = Ping::new();
+                    ping.set_timestamp(ping.get_timestamp());
+                    self.stream.write(MumblePacket::Ping(ping)).await?;
+                }
+            }
+            _ => println!("unimplemented!")
+        }
+        Ok(())
+    }
+}
+
+impl From<crate::protocol::Error> for Error {
+    fn from(_: crate::protocol::Error) -> Self {
+        Error::ConnectionSetupError
+    }
+}
+

src/db.rs

@@ -1,19 +1,72 @@
+use serde::{Deserialize, Serialize};
+
+const USER_TREE_NAME: &[u8] = b"users";
+const CHANNEL_TREE_NAME: &[u8] = b"channels";
+const ROOT_CHANNEL_KEY: &[u8] = &0_u64.to_be_bytes();
+
 pub struct Db {
     db: sled::Db,
     users: sled::Tree,
     channels: sled::Tree,
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct User {
+    pub username: String,
+    pub channel_id: u32,
+    pub is_connected: bool,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Channel {
+    pub id: u32,
+    pub name: String,
+}
+
 impl Db {
     pub fn open(path_to_db_file: &str) -> Self {
         let db = sled::open(path_to_db_file).expect("Unable to open database");
         let users = db.open_tree(USER_TREE_NAME).unwrap();
         let channels = db.open_tree(CHANNEL_TREE_NAME).unwrap();
 
+        let root_channel = bincode::serialize(&Channel {
+            id: 0,
+            name: "Root".to_string(),
+        }).unwrap();
+        channels.compare_and_swap(
+            ROOT_CHANNEL_KEY,
+            Option::<&[u8]>::None,
+            Some(root_channel))
+            .unwrap().unwrap();
+
         Db {
             db,
             users,
             channels,
         }
     }
+
+    pub async fn add_new_user(&self, user: User) {
+        let id = self.users.len().to_be_bytes();
+
+        self.users.insert(
+            id,
+            bincode::serialize(&user).unwrap(),
+        ).unwrap();
+
+        self.users.flush_async().await.unwrap();
+    }
+
+    pub fn get_channels(&self) -> Vec<Channel> {
+        self.channels.iter().values()
+            .map(|channel| bincode::deserialize(&channel.unwrap()).unwrap())
+            .collect()
+    }
+
+    pub fn get_connected_users(&self) -> Vec<User> {
+        self.users.iter().values()
+            .map(|user| bincode::deserialize(&user.unwrap()).unwrap())
+            .filter(|user: &User| user.is_connected)
+            .collect()
+    }
 }

src/main.rs

@@ -8,6 +8,7 @@ use tokio_rustls::rustls::{Certificate, internal::pemfile, PrivateKey};
 mod server;
 mod proto;
 mod protocol;
+mod connection;
 mod db;
 
 fn main() {

src/protocol.rs

@@ -6,6 +6,8 @@ use crate::proto::mumble::{Version, Authenticate, Ping, Reject, ServerSync,
                            ContextActionModify, ContextAction, UserList, VoiceTarget, PermissionQuery,
                            CodecVersion, UserStats, RequestBlob, ServerConfig, SuggestConfig};
 
+pub const MUMBLE_PROTOCOL_VERSION: u32 = 0b0000_0001_0011_0100;
+
 const VERSION: u16 = 0;
 const UDP_TUNNEL: u16 = 1;
 const AUTHENTICATE: u16 = 2;