merge main

Reviewed-on: lavina/lavina#12

.gitea/workflows/test-pr.yml

@@ -16,4 +16,3 @@ jobs:
       uses: https://github.com/actions-rs/cargo@v1
       with:
         command: test
-        args: --workspace -- --skip projections::irc

Cargo.lock

@@ -854,24 +854,31 @@ dependencies = [
  "futures-util",
  "http-body-util",
  "hyper 1.0.0-rc.3",
- "lazy_static",
+ "lavina-core",
  "mgmt-api",
- "nom",
  "nonempty",
+ "projection-irc",
+ "projection-xmpp",
  "prometheus",
- "proto-irc",
- "quick-xml",
  "regex",
  "reqwest",
- "rustls-pemfile",
  "serde",
  "serde_json",
- "sqlx",
  "tokio",
- "tokio-rustls",
  "tracing",
  "tracing-subscriber",
- "uuid",
+]
+
+[[package]]
+name = "lavina-core"
+version = "0.0.1-dev"
+dependencies = [
+ "anyhow",
+ "prometheus",
+ "serde",
+ "sqlx",
+ "tokio",
+ "tracing",
 ]
 
 [[package]]
@@ -1241,6 +1248,40 @@ dependencies = [
  "yansi",
 ]
 
+[[package]]
+name = "projection-irc"
+version = "0.0.1-dev"
+dependencies = [
+ "anyhow",
+ "futures-util",
+ "lavina-core",
+ "nonempty",
+ "prometheus",
+ "proto-irc",
+ "serde",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "projection-xmpp"
+version = "0.0.1-dev"
+dependencies = [
+ "anyhow",
+ "derive_more",
+ "futures-util",
+ "lavina-core",
+ "prometheus",
+ "proto-xmpp",
+ "quick-xml",
+ "rustls-pemfile",
+ "serde",
+ "tokio",
+ "tokio-rustls",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "prometheus"
 version = "0.13.3"
@@ -1267,6 +1308,19 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "proto-xmpp"
+version = "0.0.1-dev"
+dependencies = [
+ "anyhow",
+ "base64",
+ "derive_more",
+ "lazy_static",
+ "quick-xml",
+ "regex",
+ "tokio",
+]
+
 [[package]]
 name = "quick-xml"
 version = "0.30.0"

Cargo.toml

@@ -1,7 +1,10 @@
 [workspace]
 members = [
     ".",
+    "crates/lavina-core",
     "crates/proto-irc",
+    "crates/projection-irc",
+    "crates/proto-xmpp",
     "crates/mgmt-api",
 ]
 
@@ -15,8 +18,16 @@ tokio = { version = "1.24.1", features = ["full"] } # async runtime
 futures-util = "0.3.25"
 anyhow = "1.0.68" # error utils
 nonempty = "0.8.1"
+quick-xml = { version = "0.30.0", features = ["async-tokio"] }
+lazy_static = "1.4.0"
+regex = "1.7.1"
+derive_more = "0.99.17"
 clap = { version = "4.4.4", features = ["derive"] }
 serde = { version = "1.0.152", features = ["rc", "serde_derive"] }
+tracing = "0.1.37" # logging & tracing api
+prometheus = { version = "0.13.3", default-features = false }
+base64 = "0.21.3"
+lavina-core = { path = "crates/lavina-core" }
 
 [package]
 name = "lavina"
@@ -32,21 +43,15 @@ http-body-util = "0.1.0-rc.3"
 serde.workspace = true
 serde_json = "1.0.93"
 tokio.workspace = true
-tracing = "0.1.37" # logging & tracing api
+tracing.workspace = true
 tracing-subscriber = "0.3.16"
 futures-util.workspace = true
-prometheus = { version = "0.13.3", default-features = false }
+prometheus.workspace = true
-regex = "1.7.1"
-lazy_static = "1.4.0"
-nom.workspace = true
 nonempty.workspace = true
-tokio-rustls = "0.24.1"
+derive_more.workspace = true
-rustls-pemfile = "1.0.2"
+lavina-core.workspace = true
-quick-xml = { version = "0.30.0", features = ["async-tokio"] }
+projection-irc = { path = "crates/projection-irc" }
-derive_more = "0.99.17"
+projection-xmpp = { path = "crates/projection-xmpp" }
-uuid = { version = "1.3.0", features = ["v4"] }
-sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "migrate"] }
-proto-irc = { path = "crates/proto-irc" }
 mgmt-api = { path = "crates/mgmt-api" }
 clap.workspace = true
 

crates/README.md

@@ -0,0 +1,30 @@
+## Dependency diagram of the project
+
+```mermaid
+graph TD;
+    lavina-->mgmt-api;
+    lavina-->projection-irc;
+    lavina-->projection-xmpp;
+    lavina-->lavina-core;
+
+    projection-irc-->proto-irc;
+    projection-irc-->lavina-core;
+    
+    projection-xmpp-->proto-xmpp;
+    projection-xmpp-->lavina-core;
+
+    sim-irc-->proto-irc;
+    sim-irc-->mgmt-api;
+
+    sim-xmpp-->proto-xmpp;
+    sim-xmpp-->mgmt-api;
+
+    workspace-->lavina;
+    workspace-->sim-irc;
+    workspace-->sim-xmpp;
+```
+
+A few rules:
+- Only projections should be direct deps of `lavina`, there is no need to depend on `proto-*` crates.
+- On the other hand, projections should not be dependencies of `sim-*` crates.
+- `lavina-core` does not depend on protocol-specific crates.

crates/lavina-core/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "lavina-core"
+edition = "2021"
+version.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "migrate"] }
+serde.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+prometheus.workspace = true

crates/lavina-core/src/lib.rs

@@ -1,4 +1,8 @@
 //! Domain definitions and implementation of common chat logic.
 pub mod player;
+pub mod prelude;
 pub mod repo;
 pub mod room;
+pub mod terminator;
+
+mod table;

crates/lavina-core/src/player.rs

@@ -19,11 +19,9 @@ use tokio::{
     task::JoinHandle,
 };
 
-use crate::{
+use crate::prelude::*;
-    core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry},
+use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
-    prelude::*,
+use crate::table::{AnonTable, Key as AnonKey};
-    util::table::{AnonTable, Key as AnonKey},
-};
 
 /// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@@ -63,9 +61,7 @@ impl PlayerConnection {
     }
 
     pub async fn join_room(&mut self, room_id: RoomId) -> Result<JoinResult> {
-        self.player_handle
+        self.player_handle.join_room(room_id, self.connection_id.clone()).await
-            .join_room(room_id, self.connection_id.clone())
-            .await
     }
 
     pub async fn change_topic(&mut self, room_id: RoomId, new_topic: Str) -> Result<()> {
@@ -100,9 +96,7 @@ impl PlayerConnection {
 
     pub async fn get_rooms(&self) -> Result<Vec<RoomInfo>> {
         let (promise, deferred) = oneshot();
-        self.player_handle
+        self.player_handle.send(PlayerCommand::GetRooms(promise)).await;
-            .send(PlayerCommand::GetRooms(promise))
-            .await;
         Ok(deferred.await?)
     }
 }
@@ -126,27 +120,14 @@ impl PlayerHandle {
         }
     }
 
-    pub async fn send_message(
+    pub async fn send_message(&self, room_id: RoomId, connection_id: ConnectionId, body: Str) -> Result<()> {
-        &self,
-        room_id: RoomId,
-        connection_id: ConnectionId,
-        body: Str,
-    ) -> Result<()> {
         let (promise, deferred) = oneshot();
-        let cmd = Cmd::SendMessage {
+        let cmd = Cmd::SendMessage { room_id, body, promise };
-            room_id,
-            body,
-            promise,
-        };
         let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
         Ok(deferred.await?)
     }
 
-    pub async fn join_room(
+    pub async fn join_room(&self, room_id: RoomId, connection_id: ConnectionId) -> Result<JoinResult> {
-        &self,
-        room_id: RoomId,
-        connection_id: ConnectionId,
-    ) -> Result<JoinResult> {
         let (promise, deferred) = oneshot();
         let cmd = Cmd::JoinRoom { room_id, promise };
         let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
@@ -230,12 +211,8 @@ pub enum Updates {
 #[derive(Clone)]
 pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
 impl PlayerRegistry {
-    pub fn empty(
+    pub fn empty(room_registry: RoomRegistry, metrics: &mut MetricsRegistry) -> Result<PlayerRegistry> {
-        room_registry: RoomRegistry,
+        let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?;
-        metrics: &mut MetricsRegistry,
-    ) -> Result<PlayerRegistry> {
-        let metric_active_players =
-            IntGauge::new("chat_players_active", "Number of alive player actors")?;
         metrics.register(Box::new(metric_active_players.clone()))?;
         let inner = PlayerRegistryInner {
             room_registry,
@@ -375,8 +352,7 @@ impl Player {
                         return;
                     }
                 };
-                room.subscribe(self.player_id.clone(), self.handle.clone())
+                room.subscribe(self.player_id.clone(), self.handle.clone()).await;
-                    .await;
                 self.my_rooms.insert(room_id.clone(), room.clone());
                 let room_info = room.get_room_info().await;
                 let _ = promise.send(JoinResult::Success(room_info));
@@ -399,15 +375,10 @@ impl Player {
                 };
                 self.broadcast_update(update, connection_id).await;
             }
-            Cmd::SendMessage {
+            Cmd::SendMessage { room_id, body, promise } => {
-                room_id,
-                body,
-                promise,
-            } => {
                 let room = self.rooms.get_room(&room_id).await;
                 if let Some(room) = room {
-                    room.send_message(self.player_id.clone(), body.clone())
+                    room.send_message(self.player_id.clone(), body.clone()).await;
-                        .await;
                 } else {
                     tracing::info!("no room found");
                 }
@@ -426,8 +397,7 @@ impl Player {
             } => {
                 let room = self.rooms.get_room(&room_id).await;
                 if let Some(mut room) = room {
-                    room.set_topic(self.player_id.clone(), new_topic.clone())
+                    room.set_topic(self.player_id.clone(), new_topic.clone()).await;
-                        .await;
                 } else {
                     tracing::info!("no room found");
                 }

crates/lavina-core/src/prelude.rs

@@ -15,11 +15,3 @@ pub type Str = std::sync::Arc<str>;
 pub fn fail(msg: &str) -> anyhow::Error {
     anyhow::Error::msg(msg.to_owned())
 }
-
-macro_rules! ffail {
-    ($($arg:tt)*) => {
-        fail(&format!($($arg)*))
-    };
-}
-
-pub(crate) use ffail;

crates/lavina-core/src/repo/mod.rs

@@ -93,10 +93,10 @@ impl Storage {
         Ok(())
     }
 
-    pub async fn close(mut self) -> Result<()> {
+    pub async fn close(self) -> Result<()> {
         let res = match Arc::try_unwrap(self.conn) {
             Ok(e) => e,
-            Err(e) => return Err(fail("failed to acquire DB ownership on shutdown")),
+            Err(_) => return Err(fail("failed to acquire DB ownership on shutdown")),
         };
         let res = res.into_inner();
         res.close().await?;

crates/lavina-core/src/room.rs

@@ -1,17 +1,13 @@
 //! Domain of rooms — chats with multiple participants.
-use std::{
+use std::{collections::HashMap, hash::Hash, sync::Arc};
-    collections::HashMap,
-    hash::Hash,
-    sync::Arc,
-};
 
 use prometheus::{IntGauge, Registry as MetricRegistry};
 use serde::Serialize;
 use tokio::sync::RwLock as AsyncRwLock;
 
-use crate::core::player::{PlayerHandle, PlayerId, Updates};
+use crate::player::{PlayerHandle, PlayerId, Updates};
-use crate::core::repo::Storage;
 use crate::prelude::*;
+use crate::repo::Storage;
 
 /// Opaque room id
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@@ -20,9 +16,7 @@ impl RoomId {
     pub fn from(str: impl Into<Str>) -> Result<RoomId> {
         let bytes = str.into();
         if bytes.len() > 32 {
-            return Err(anyhow::Error::msg(
+            return Err(anyhow::Error::msg("Room name cannot be longer than 32 symbols"));
-                "Room name cannot be longer than 32 symbols",
-            ));
         }
         if bytes.contains(' ') {
             return Err(anyhow::Error::msg("Room name cannot contain spaces"));
@@ -42,8 +36,7 @@ impl RoomId {
 pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
 impl RoomRegistry {
     pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> {
-        let metric_active_rooms =
+        let metric_active_rooms = IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
-            IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
         metrics.register(Box::new(metric_active_rooms.clone()))?;
         let inner = RoomRegistryInner {
             rooms: HashMap::new(),
@@ -140,7 +133,7 @@ impl RoomHandle {
 
     pub async fn send_message(&self, player_id: PlayerId, body: Str) {
         let mut lock = self.0.write().await;
-        let res =  lock.send_message(player_id, body).await;
+        let res = lock.send_message(player_id, body).await;
         if let Err(err) = res {
             log::warn!("Failed to send message: {err:?}");
         }
@@ -150,11 +143,7 @@ impl RoomHandle {
         let lock = self.0.read().await;
         RoomInfo {
             id: lock.room_id.clone(),
-            members: lock
+            members: lock.subscriptions.keys().map(|x| x.clone()).collect::<Vec<_>>(),
-                .subscriptions
-                .keys()
-                .map(|x| x.clone())
-                .collect::<Vec<_>>(),
             topic: lock.topic.clone(),
         }
     }
@@ -191,7 +180,9 @@ impl Room {
 
     async fn send_message(&mut self, author_id: PlayerId, body: Str) -> Result<()> {
         tracing::info!("Adding a message to room");
-        self.storage.insert_message(self.storage_id, self.message_count, &body).await?;
+        self.storage
+            .insert_message(self.storage_id, self.message_count, &body)
+            .await?;
         self.message_count += 1;
         let update = Updates::NewMessage {
             room_id: self.room_id.clone(),

crates/lavina-core/src/terminator.rs

@@ -1,11 +1,5 @@
 use crate::prelude::*;
 
-pub mod table;
-pub mod telemetry;
-#[cfg(test)]
-pub mod testkit;
-pub mod xml;
-
 pub struct Terminator {
     signal: Promise<()>,
     completion: JoinHandle<Result<()>>,

crates/projection-irc/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "projection-irc"
+edition = "2021"
+version.workspace = true
+
+[dependencies]
+lavina-core.workspace = true
+tracing.workspace = true
+anyhow.workspace = true
+serde.workspace = true
+tokio.workspace = true
+prometheus.workspace = true
+futures-util.workspace = true
+
+nonempty.workspace = true
+proto-irc = { path = "../proto-irc" }

crates/projection-irc/src/lib.rs

@@ -1,29 +1,26 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
 
+use anyhow::{anyhow, Result};
 use futures_util::future::join_all;
 use nonempty::nonempty;
 use nonempty::NonEmpty;
 use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
-
 use serde::Deserialize;
 use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
 use tokio::net::tcp::{ReadHalf, WriteHalf};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::mpsc::channel;
 
-use crate::core::player::*;
+use lavina_core::player::*;
-use crate::core::repo::Storage;
+use lavina_core::prelude::*;
-use crate::core::room::{RoomId, RoomInfo, RoomRegistry};
+use lavina_core::repo::Storage;
-use crate::prelude::*;
+use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
+use lavina_core::terminator::Terminator;
 use proto_irc::client::{client_message, ClientMessage};
 use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
+use proto_irc::user::PrefixedNick;
 use proto_irc::{Chan, Recipient};
-use proto_irc::user::{Prefix, PrefixedNick};
-use crate::util::Terminator;
-
-#[cfg(test)]
-mod test;
 
 #[derive(Deserialize, Debug, Clone)]
 pub struct ServerConfig {
@@ -70,14 +67,12 @@ async fn handle_socket(
     .await?;
     writer.flush().await?;
 
-    let registered_user: Result<RegisteredUser> =
+    let registered_user: Result<RegisteredUser> = handle_registration(&mut reader, &mut writer, &mut storage).await;
-        handle_registration(&mut reader, &mut writer, &mut storage).await;
 
     match registered_user {
         Ok(user) => {
             log::debug!("User registered");
-            handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user)
+            handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user).await?;
-                .await?;
         }
         Err(_) => {
             log::debug!("Registration failed");
@@ -159,23 +154,22 @@ async fn handle_registration<'a>(
         buffer.clear();
     }?;
 
-
     let stored_user = storage.retrieve_user_by_name(&*user.nickname).await?;
 
     let stored_user = match stored_user {
         Some(u) => u,
         None => {
             log::info!("User '{}' not found", user.nickname);
-            return Err(fail("no user found"));
+            return Err(anyhow!("no user found"));
         }
     };
     if stored_user.password.is_none() {
         log::info!("Password not defined for user '{}'", user.nickname);
-        return Err(fail("password is not defined"));
+        return Err(anyhow!("password is not defined"));
     }
     if stored_user.password.as_deref() != pass.as_deref() {
         log::info!("Incorrect password supplied for user '{}'", user.nickname);
-        return Err(fail("passwords do not match"));
+        return Err(anyhow!("passwords do not match"));
     }
     // TODO properly implement session temination
 
@@ -250,14 +244,7 @@ async fn handle_registered_socket<'a>(
 
     let rooms_list = connection.get_rooms().await?;
     for room in &rooms_list {
-        produce_on_join_cmd_messages(
+        produce_on_join_cmd_messages(&config, &user, &Chan::Global(room.id.as_inner().clone()), room, writer).await?;
-            &config,
-            &user,
-            &Chan::Global(room.id.as_inner().clone()),
-            room,
-            writer,
-        )
-        .await?;
     }
 
     writer.flush().await?;
@@ -314,10 +301,7 @@ async fn handle_update(
 ) -> Result<()> {
     log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
     match update {
-        Updates::RoomJoined {
+        Updates::RoomJoined { new_member_id, room_id } => {
-            new_member_id,
-            room_id,
-        } => {
             if player_id == &new_member_id {
                 if let Some(room) = rooms.get_room(&room_id).await {
                     let room_info = room.get_room_info().await;
@@ -438,16 +422,14 @@ async fn handle_incoming_message(
                 Recipient::Chan(Chan::Global(chan)) => {
                     let room_id = RoomId::from(chan)?;
                     user_handle.send_message(room_id, body).await?;
-                },
+                }
                 _ => log::warn!("Unsupported target type"),
             },
             ClientMessage::Topic { chan, topic } => {
                 match chan {
                     Chan::Global(chan) => {
                         let room_id = RoomId::from(chan)?;
-                        user_handle
+                        user_handle.change_topic(room_id.clone(), topic.clone()).await?;
-                            .change_topic(room_id.clone(), topic.clone())
-                            .await?;
                         ServerMessage {
                             tags: vec![],
                             sender: Some(config.server_name.clone()),
@@ -569,11 +551,7 @@ async fn handle_incoming_message(
     Ok(HandleResult::Continue)
 }
 
-fn user_to_who_msg(
+fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_user_nickname: &Str) -> ServerMessageBody {
-    config: &ServerConfig,
-    requestor: &RegisteredUser,
-    target_user_nickname: &Str,
-) -> ServerMessageBody {
     // Username is equal to nickname
     let username = format!("~{target_user_nickname}").into();
 
@@ -674,8 +652,13 @@ async fn produce_on_join_cmd_messages(
     }
     .write_async(writer)
     .await?;
-    let prefixed_members: Vec<PrefixedNick> = room_info.members.iter().map(|member| PrefixedNick::from_str(member.clone().into_inner())).collect();
+    let prefixed_members: Vec<PrefixedNick> = room_info
-    let non_empty_members: NonEmpty<PrefixedNick> = NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
+        .members
+        .iter()
+        .map(|member| PrefixedNick::from_str(member.clone().into_inner()))
+        .collect();
+    let non_empty_members: NonEmpty<PrefixedNick> =
+        NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
 
     ServerMessage {
         tags: vec![],
@@ -710,12 +693,8 @@ pub async fn launch(
 ) -> Result<Terminator> {
     log::info!("Starting IRC projection");
     let (stopped_tx, mut stopped_rx) = channel(32);
-    let current_connections =
+    let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
-        IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
+    let total_connections = IntCounter::new("irc_total_connections", "Total number of opened connections")?;
-    let total_connections = IntCounter::new(
-        "irc_total_connections",
-        "Total number of opened connections",
-    )?;
     metrics.register(Box::new(current_connections.clone()))?;
     metrics.register(Box::new(total_connections.clone()))?;
 
@@ -780,7 +759,8 @@ pub async fn launch(
                     log::warn!("IRC connection to {socket_addr} finished with error: {err}")
                 }
             }
-        })).await;
+        }))
+        .await;
         log::info!("Stopped IRC projection");
         Ok(())
     });

crates/projection-xmpp/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "projection-xmpp"
+edition = "2021"
+version.workspace = true
+
+[dependencies]
+lavina-core.workspace = true
+tracing.workspace = true
+anyhow.workspace = true
+serde.workspace = true
+tokio.workspace = true
+prometheus.workspace = true
+futures-util.workspace = true
+
+quick-xml.workspace = true
+proto-xmpp = { path = "../proto-xmpp" }
+uuid = { version = "1.3.0", features = ["v4"] }
+tokio-rustls = "0.24.1"
+rustls-pemfile = "1.0.2"
+derive_more.workspace = true

crates/projection-xmpp/src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(generators, generator_trait, type_alias_impl_trait, impl_trait_in_assoc_type)]
+
 mod proto;
 
 use std::collections::HashMap;
@@ -7,6 +9,7 @@ use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use anyhow::anyhow;
 use futures_util::future::join_all;
 use prometheus::Registry as MetricsRegistry;
 use quick_xml::events::{BytesDecl, Event};
@@ -19,18 +22,19 @@ use tokio::sync::mpsc::channel;
 use tokio_rustls::rustls::{Certificate, PrivateKey};
 use tokio_rustls::TlsAcceptor;
 
-use crate::core::player::{PlayerConnection, PlayerId, PlayerRegistry};
+use lavina_core::player::{PlayerConnection, PlayerId, PlayerRegistry};
-use crate::core::room::{RoomId, RoomRegistry};
+use lavina_core::prelude::*;
-use crate::prelude::*;
+use lavina_core::room::{RoomId, RoomRegistry};
-use crate::protos::xmpp;
+use lavina_core::terminator::Terminator;
-use crate::protos::xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
+use lavina_core::repo::Storage;
-use crate::protos::xmpp::client::{Iq, Message, MessageType, Presence};
+use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
-use crate::protos::xmpp::disco::*;
+use proto_xmpp::client::{Iq, Message, MessageType, Presence};
-use crate::protos::xmpp::roster::RosterQuery;
+use proto_xmpp::disco::*;
-use crate::protos::xmpp::session::Session;
+use proto_xmpp::roster::RosterQuery;
-use crate::protos::xmpp::stream::*;
+use proto_xmpp::sasl::AuthBody;
-use crate::util::xml::{Continuation, FromXml, Parser, ToXml};
+use proto_xmpp::session::Session;
-use crate::util::Terminator;
+use proto_xmpp::stream::*;
+use proto_xmpp::xml::{Continuation, FromXml, Parser, ToXml};
 
 use self::proto::{ClientPacket, IqClientBody};
 
@@ -58,6 +62,7 @@ pub async fn launch(
     players: PlayerRegistry,
     rooms: RoomRegistry,
     metrics: MetricsRegistry,
+    storage: Storage,
 ) -> Result<Terminator> {
     log::info!("Starting XMPP projection");
 
@@ -97,11 +102,12 @@ pub async fn launch(
                             }
                             let players = players.clone();
                             let rooms = rooms.clone();
+                            let storage = storage.clone();
                             let terminator = Terminator::spawn(|termination| {
                                 let stopped_tx = stopped_tx.clone();
                                 let loaded_config = loaded_config.clone();
                                 async move {
-                                    match handle_socket(loaded_config, stream, &socket_addr, players, rooms, termination).await {
+                                    match handle_socket(loaded_config, stream, &socket_addr, players, rooms, storage, termination).await {
                                         Ok(_) => log::info!("Connection terminated"),
                                         Err(err) => log::warn!("Connection failed: {err}"),
                                     }
@@ -118,21 +124,15 @@ pub async fn launch(
             }
         }
         log::info!("Stopping XMPP projection");
-        join_all(
+        join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
-            actors
+            log::debug!("Stopping XMPP connection at {socket_addr}");
-                .into_iter()
+            match terminator.terminate().await {
-                .map(|(socket_addr, terminator)| async move {
+                Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
-                    log::debug!("Stopping XMPP connection at {socket_addr}");
+                Err(err) => {
-                    match terminator.terminate().await {
+                    log::warn!("XMPP connection to {socket_addr} finished with error: {err}")
-                        Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
+                }
-                        Err(err) => {
+            }
-                            log::warn!(
+        }))
-                                "XMPP connection to {socket_addr} finished with error: {err}"
-                            )
-                        }
-                    }
-                }),
-        )
         .await;
         log::info!("Stopped XMPP projection");
         Ok(())
@@ -147,6 +147,7 @@ async fn handle_socket(
     socket_addr: &SocketAddr,
     mut players: PlayerRegistry,
     rooms: RoomRegistry,
+    storage: Storage,
     termination: Deferred<()>, // TODO use it to stop the connection gracefully
 ) -> Result<()> {
     log::info!("Received an XMPP connection from {socket_addr}");
@@ -171,11 +172,10 @@ async fn handle_socket(
     let mut xml_reader = NsReader::from_reader(BufReader::new(a));
     let mut xml_writer = Writer::new(b);
 
-    let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf).await?;
+    let mut storage = storage.clone();
+    let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &mut storage).await?;
     log::debug!("User authenticated");
-    let mut connection = players
+    let mut connection = players.connect_to_player(authenticated.player_id.clone()).await;
-        .connect_to_player(authenticated.player_id.clone())
-        .await;
     socket_final(
         &mut xml_reader,
         &mut xml_writer,
@@ -197,7 +197,7 @@ async fn socket_force_tls(
     writer: &mut (impl AsyncWrite + Unpin),
     reader_buf: &mut Vec<u8>,
 ) -> Result<()> {
-    use crate::protos::xmpp::tls::*;
+    use proto_xmpp::tls::*;
     let xml_reader = &mut NsReader::from_reader(reader);
     let xml_writer = &mut Writer::new(writer);
     read_xml_header(xml_reader, reader_buf).await?;
@@ -230,6 +230,7 @@ async fn socket_auth(
     xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>,
     xml_writer: &mut Writer<(impl AsyncWrite + Unpin)>,
     reader_buf: &mut Vec<u8>,
+    storage: &mut Storage,
 ) -> Result<Authenticated> {
     read_xml_header(xml_reader, reader_buf).await?;
     let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
@@ -254,16 +255,41 @@ async fn socket_auth(
     .await?;
     xml_writer.get_mut().flush().await?;
 
-    let _ = xmpp::sasl::Auth::parse(xml_reader, reader_buf).await?;
+    let auth: proto_xmpp::sasl::Auth = proto_xmpp::sasl::Auth::parse(xml_reader, reader_buf).await?;
-    xmpp::sasl::Success.write_xml(xml_writer).await?;
+    proto_xmpp::sasl::Success.write_xml(xml_writer).await?;
 
-    let name: Str = "darova".into();
+    let auth_body = AuthBody::from_str(auth.body);
-    Ok(Authenticated {
+
-        player_id: PlayerId::from("darova")?,
+    match auth_body {
-        xmpp_name: Name(name.clone()),
+        Ok(logopass) => {
-        xmpp_resource: Resource(name.clone()),
+            let name = &logopass.login;
-        xmpp_muc_name: Resource(name),
+            let stored_user = storage.retrieve_user_by_name(name).await?;
-    })
+
+            let stored_user = match stored_user {
+                Some(u) => u,
+                None => {
+                    log::info!("User '{}' not found", name);
+                    return Err(fail("no user found"));
+                }
+            };
+            if stored_user.password.is_none() {
+                log::info!("Password not defined for user '{}'", name);
+                return Err(fail("password is not defined"));
+            }
+            if stored_user.password.as_deref() != Some(&logopass.password) {
+                log::info!("Incorrect password supplied for user '{}'", name);
+                return Err(fail("passwords do not match"));
+            }
+
+            Ok(Authenticated {
+                player_id: PlayerId::from("darova")?,
+                xmpp_name: Name(name.to_string().into()),
+                xmpp_resource: Resource(name.to_string().into()),
+                xmpp_muc_name: Resource(name.to_string().into()),
+            })
+        }
+        Err(e) => Err(e)
+    }
 }
 
 async fn socket_final(
@@ -333,7 +359,7 @@ async fn socket_final(
             update = user_handle.receiver.recv() => {
                 if let Some(update) = update {
                     match update {
-                        crate::core::player::Updates::NewMessage { room_id, author_id, body } => {
+                        lavina_core::player::Updates::NewMessage { room_id, author_id, body } => {
                             Message {
                                 to: Some(Jid {
                                     name: Some(authenticated.xmpp_name.clone()),
@@ -346,7 +372,7 @@ async fn socket_final(
                                     resource: Some(Resource(author_id.into_inner().into())),
                                 }),
                                 id: None,
-                                r#type: xmpp::client::MessageType::Groupchat,
+                                r#type: proto_xmpp::client::MessageType::Groupchat,
                                 lang: None,
                                 subject: None,
                                 body: body.into(),
@@ -397,10 +423,7 @@ async fn handle_packet(
             {
                 if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat {
                     user_handle
-                        .send_message(
+                        .send_message(RoomId::from(name.0.clone())?, m.body.clone().into())
-                            RoomId::from(name.0.clone())?,
-                            m.body.clone().into(),
-                        )
                         .await?;
                     Message {
                         to: Some(Jid {
@@ -414,7 +437,7 @@ async fn handle_packet(
                             resource: Some(user.xmpp_muc_name.clone()),
                         }),
                         id: m.id,
-                        r#type: xmpp::client::MessageType::Groupchat,
+                        r#type: proto_xmpp::client::MessageType::Groupchat,
                         lang: None,
                         subject: None,
                         body: m.body.clone(),
@@ -449,9 +472,7 @@ async fn handle_packet(
                 resource: Some(resource),
             }) = p.to
             {
-                let a = user_handle
+                let a = user_handle.join_room(RoomId::from(name.0.clone())?).await?;
-                    .join_room(RoomId::from(name.0.clone())?)
-                    .await?;
                 Presence::<()> {
                     to: Some(Jid {
                         name: Some(user.xmpp_name.clone()),
@@ -485,7 +506,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: None,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Result,
+                r#type: proto_xmpp::client::IqType::Result,
                 body: BindResponse(Jid {
                     name: Some(Name("darova".into())),
                     server: Server("localhost".into()),
@@ -499,7 +520,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: None,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Result,
+                r#type: proto_xmpp::client::IqType::Result,
                 body: Session,
             };
             req.serialize(output);
@@ -509,7 +530,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: None,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Result,
+                r#type: proto_xmpp::client::IqType::Result,
                 body: RosterQuery,
             };
             req.serialize(output);
@@ -520,7 +541,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: iq.to,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Result,
+                r#type: proto_xmpp::client::IqType::Result,
                 body: response,
             };
             req.serialize(output);
@@ -531,7 +552,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: iq.to,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Result,
+                r#type: proto_xmpp::client::IqType::Result,
                 body: response,
             };
             req.serialize(output);
@@ -541,7 +562,7 @@ async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms
                 from: None,
                 id: iq.id,
                 to: None,
-                r#type: xmpp::client::IqType::Error,
+                r#type: proto_xmpp::client::IqType::Error,
                 body: (),
             };
             req.serialize(output);
@@ -634,13 +655,13 @@ async fn read_xml_header(
             if &*encoding == b"UTF-8" {
                 Ok(())
             } else {
-                Err(fail(format!("Unsupported encoding: {encoding:?}").as_str()))
+                Err(anyhow!("Unsupported encoding: {encoding:?}"))
             }
         } else {
             // Err(fail("No XML encoding provided"))
             Ok(())
         }
     } else {
-        Err(fail("Expected XML header"))
+        Err(anyhow!("Expected XML header"))
     }
 }

crates/projection-xmpp/src/proto.rs

@@ -1,15 +1,15 @@
+use anyhow::anyhow;
 use derive_more::From;
 use quick_xml::events::Event;
 use quick_xml::name::{Namespace, ResolveResult};
 
-use crate::protos::xmpp::bind::BindRequest;
+use lavina_core::prelude::*;
-use crate::protos::xmpp::client::{Iq, Message, Presence};
+use proto_xmpp::bind::BindRequest;
-use crate::protos::xmpp::disco::{InfoQuery, ItemQuery};
+use proto_xmpp::client::{Iq, Message, Presence};
-use crate::protos::xmpp::roster::RosterQuery;
+use proto_xmpp::disco::{InfoQuery, ItemQuery};
-use crate::protos::xmpp::session::Session;
+use proto_xmpp::roster::RosterQuery;
-use crate::util::xml::*;
+use proto_xmpp::session::Session;
-
+use proto_xmpp::xml::*;
-use crate::prelude::*;
 
 #[derive(PartialEq, Eq, Debug, From)]
 pub enum IqClientBody {
@@ -29,7 +29,7 @@ impl FromXml for IqClientBody {
             let bytes = match event {
                 Event::Start(bytes) => bytes,
                 Event::Empty(bytes) => bytes,
-                _ => return Err(ffail!("Unexpected XML event: {event:?}")),
+                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
             };
             let name = bytes.name();
             match_parser!(name, namespace, event;
@@ -67,7 +67,7 @@ impl FromXml for ClientPacket {
                         Presence::<Ignore>,
                         Message,
                         {
-                            Err(ffail!(
+                            Err(anyhow!(
                                 "Unexpected XML event of name {:?} in namespace {:?}",
                                 name,
                                 namespace
@@ -80,11 +80,11 @@ impl FromXml for ClientPacket {
                     if name.local_name().as_ref() == b"stream" {
                         return Ok(ClientPacket::StreamEnd);
                     } else {
-                        return Err(ffail!("Unexpected XML event: {event:?}"));
+                        return Err(anyhow!("Unexpected XML event: {event:?}"));
                     }
                 }
                 _ => {
-                    return Err(ffail!("Unexpected XML event: {event:?}"));
+                    return Err(anyhow!("Unexpected XML event: {event:?}"));
                 }
             }
         }

crates/proto-xmpp/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "proto-xmpp"
+edition = "2021"
+version.workspace = true
+
+[dependencies]
+quick-xml.workspace = true
+lazy_static.workspace = true
+regex.workspace = true
+anyhow.workspace = true
+tokio.workspace = true
+derive_more.workspace = true
+base64.workspace = true

crates/proto-xmpp/src/bind.rs

@@ -1,11 +1,11 @@
 use std::fmt::Display;
 
-use nom::AsBytes;
+use anyhow::{anyhow, Result};
 use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
 use quick_xml::name::{Namespace, ResolveResult};
 
 use crate::prelude::*;
-use crate::util::xml::*;
+use crate::xml::*;
 
 pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
 
@@ -49,7 +49,7 @@ impl Jid {
         }
         let m = RE
             .captures(i)
-            .ok_or(ffail!("Incorrectly format jid: {i}"))?;
+            .ok_or(anyhow!("Incorrectly format jid: {i}"))?;
 
         let name = m.get(2).map(|name| Name(name.as_str().into()));
         let server = m.get(3).unwrap();
@@ -78,89 +78,56 @@ impl Jid {
 #[derive(PartialEq, Eq, Debug)]
 pub struct BindRequest(pub Resource);
 
-pub struct BindRequestParser(BindRequestParserInner);
-
-enum BindRequestParserInner {
-    Initial,
-    /// Consumed <bind> start and expects <resource>
-    InBind(Option<String>),
-    /// Consumed <resource> start
-    InBindResourceInitial,
-    /// Consumer <resource> start and inner text
-    InBindResourceEnd(String),
-}
-
 impl FromXmlTag for BindRequest {
     const NS: &'static str = XMLNS;
     const NAME: &'static str = "bind";
 }
 
 impl FromXml for BindRequest {
-    type P = BindRequestParser;
+    type P = impl Parser<Output = Result<Self>>;
 
     fn parse() -> Self::P {
-        BindRequestParser(BindRequestParserInner::Initial)
+        |(namespace, event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
-    }
+            let mut resource: Option<Str> = None;
-}
+            let Event::Start(bytes) = event else {
-
+                return Err(anyhow!("Unexpected XML event: {event:?}"));
-// TODO rewrite as a generator
+            };
-impl Parser for BindRequestParser {
+            if bytes.name().0 != BindRequest::NAME.as_bytes() {
-    type Output = Result<BindRequest>;
+                return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name()));
-
-    fn consume<'a>(
-        self: Self,
-        namespace: ResolveResult,
-        event: &Event<'a>,
-    ) -> Continuation<Self, Self::Output> {
-        // TODO validate tag names and namespaces
-        use BindRequestParserInner::*;
-        match self.0 {
-            Initial => {
-                let Event::Start(bytes) = event else {
-                    return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
-                };
-                if bytes.name().0 != BindRequest::NAME.as_bytes() {
-                    return Continuation::Final(Err(ffail!(
-                        "Unexpected XML tag: {:?}",
-                        bytes.name()
-                    )));
-                }
-                let ResolveResult::Bound(Namespace(ns)) = namespace else {
-                    return Continuation::Final(Err(ffail!("No namespace provided")));
-                };
-                if ns != XMLNS.as_bytes() {
-                    return Continuation::Final(Err(ffail!("Incorrect namespace")));
-                }
-                Continuation::Continue(BindRequestParser(InBind(None)))
             }
-            InBind(resource) => match event {
+            let ResolveResult::Bound(Namespace(ns)) = namespace else {
-                Event::Start(bytes) => {
+                return Err(anyhow!("No namespace provided"));
-                    Continuation::Continue(BindRequestParser(InBindResourceInitial))
+            };
-                }
+            if ns != XMLNS.as_bytes() {
-                Event::End(bytes) => {
+                return Err(anyhow!("Incorrect namespace"));
-                    let Some(resource) = resource else {
+            }
-                            return Continuation::Final(Err(ffail!("No resource was provided")));
+            loop {
+                let (namespace, event) = yield;
+                match event {
+                    Event::Start(bytes) if bytes.name().0 == b"resource" => {
+                        let (namespace, event) = yield;
+                        let Event::Text(text) = event else {
+                            return Err(anyhow!("Unexpected XML event: {event:?}"));
                         };
-                    Continuation::Final(Ok(BindRequest(Resource(resource.into()))))
+                        resource = Some(std::str::from_utf8(&*text)?.into());
+                        let (namespace, event) = yield;
+                        let Event::End(bytes) = event else {
+                            return Err(anyhow!("Unexpected XML event: {event:?}"));
+                        };
+                        if bytes.name().0 != b"resource" {
+                            return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name()));
+                        }
+                    }
+                    Event::End(bytes) if bytes.name().0 == BindRequest::NAME.as_bytes() => {
+                        break;
+                    }
+                    _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
                 }
-                _ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))),
-            },
-            InBindResourceInitial => {
-                let Event::Text(text) = event else {
-                    return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
-                };
-                let resource = match std::str::from_utf8(text.as_bytes()) {
-                    Ok(e) => e.to_string(),
-                    Err(err) => return Continuation::Final(Err(err.into())),
-                };
-                Continuation::Continue(BindRequestParser(InBindResourceEnd(resource)))
-            }
-            InBindResourceEnd(resource) => {
-                let Event::End(bytes) = event else {
-                    return Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")));
-                };
-                Continuation::Continue(BindRequestParser(InBind(Some(resource))))
             }
+            let Some(resource) = resource else {
+                return Err(anyhow!("No resource was provided"));
+            };
+            Ok(BindRequest(Resource(resource)))
         }
     }
 }

crates/proto-xmpp/src/client.rs

@@ -3,8 +3,11 @@ use quick_xml::events::attributes::Attribute;
 use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
 use quick_xml::name::{QName, ResolveResult};
 
+use anyhow::{Result, anyhow as ffail};
+
+
 use crate::prelude::*;
-use crate::util::xml::*;
+use crate::xml::*;
 
 use super::bind::Jid;
 
@@ -308,9 +311,9 @@ impl<T: FromXml> Parser for IqParser<T> {
             }
             IqParserInner::Final(state) => {
                 if let Event::End(ref bytes) = event {
-                    let id = fail_fast!(state.id.ok_or_else(|| fail("No id provided")));
+                    let id = fail_fast!(state.id.ok_or_else(|| ffail!("No id provided")));
-                    let r#type = fail_fast!(state.r#type.ok_or_else(|| fail("No type provided")));
+                    let r#type = fail_fast!(state.r#type.ok_or_else(|| ffail!("No type provided")));
-                    let body = fail_fast!(state.body.ok_or_else(|| fail("No body provided")));
+                    let body = fail_fast!(state.body.ok_or_else(|| ffail!("No body provided")));
                     Continuation::Final(Ok(Iq {
                         from: state.from,
                         id,
@@ -579,7 +582,7 @@ impl<T: ToXml> ToXml for Presence<T> {
 
 #[cfg(test)]
 mod tests {
-    use crate::protos::xmpp::bind::{BindRequest, Name, Resource, Server};
+    use crate::bind::{BindRequest, Name, Resource, Server};
 
     use super::*;
     use quick_xml::NsReader;

crates/proto-xmpp/src/disco.rs

@@ -2,8 +2,8 @@ use quick_xml::events::attributes::Attribute;
 use quick_xml::events::{BytesEnd, BytesStart, Event};
 use quick_xml::name::{QName, ResolveResult};
 
-use crate::prelude::*;
+use anyhow::{Result, anyhow as ffail};
-use crate::util::xml::*;
+use crate::xml::*;
 
 use super::bind::Jid;
 

crates/proto-xmpp/src/lib.rs

@@ -1,3 +1,10 @@
+#![feature(
+    generators,
+    generator_trait,
+    type_alias_impl_trait,
+    impl_trait_in_assoc_type
+)]
+
 pub mod bind;
 pub mod client;
 pub mod disco;
@@ -8,6 +15,8 @@ pub mod session;
 pub mod stanzaerror;
 pub mod stream;
 pub mod tls;
+mod prelude;
+pub mod xml;
 
 // Implemented as a macro instead of a fn due to borrowck limitations
 macro_rules! skip_text {
@@ -25,4 +34,4 @@ macro_rules! skip_text {
     };
 }
 
-pub(super) use skip_text;
+pub(crate) use skip_text;

crates/proto-xmpp/src/muc/mod.rs

@@ -1,8 +1,8 @@
 use quick_xml::events::Event;
 use quick_xml::name::ResolveResult;
 
-use crate::prelude::*;
+use anyhow::{anyhow, Result};
-use crate::util::xml::*;
+use crate::xml::*;
 
 pub const XMLNS: &'static str = "http://jabber.org/protocol/muc";
 
@@ -22,7 +22,7 @@ impl FromXml for History {
             let (bytes, end) = match event {
                 Event::Start(bytes) => (bytes, false),
                 Event::Empty(bytes) => (bytes, true),
-                _ => return Err(ffail!("Unexpected XML event: {event:?}")),
+                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
             };
             for attr in bytes.attributes() {
                 let attr = attr?;
@@ -51,7 +51,7 @@ impl FromXml for History {
 
             let (namespace, event) = yield;
             let Event::End(bytes) = event else {
-                return Err(ffail!("Unexpected XML event: {event:?}"));
+                return Err(anyhow!("Unexpected XML event: {event:?}"));
             };
             Ok(history)
         }
@@ -73,15 +73,15 @@ impl FromXml for Password {
     fn parse() -> Self::P {
         |(namespace, event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
             let Event::Start(bytes) = event else {
-                return Err(ffail!("Unexpected XML event: {event:?}"));
+                return Err(anyhow!("Unexpected XML event: {event:?}"));
             };
             let (namespace, event) = yield;
             let Event::Text(bytes) = event else {
-                return Err(ffail!("Unexpected XML event: {event:?}"));
+                return Err(anyhow!("Unexpected XML event: {event:?}"));
             };
             let s = std::str::from_utf8(bytes)?.to_string();
             let Event::End(bytes) = event else {
-                return Err(ffail!("Unexpected XML event: {event:?}"));
+                return Err(anyhow!("Unexpected XML event: {event:?}"));
             };
             Ok(Password(s))
         }
@@ -109,7 +109,7 @@ impl FromXml for X {
             let (bytes, end) = match event {
                 Event::Start(bytes) => (bytes, false),
                 Event::Empty(bytes) => (bytes, true),
-                _ => return Err(ffail!("Unexpected XML event: {event:?}")),
+                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
             };
             if end {
                 return Ok(res);
@@ -121,7 +121,7 @@ impl FromXml for X {
                     Event::Start(bytes) => bytes,
                     Event::Empty(bytes) => bytes,
                     Event::End(_) => break,
-                    _ => return Err(ffail!("Unexpected XML event: {event:?}")),
+                    _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
                 };
                 if bytes.name().0 == Password::NAME.as_bytes() {
                     let password = delegate_parsing!(Password, namespace, event)?;
@@ -130,7 +130,7 @@ impl FromXml for X {
                     let history = delegate_parsing!(History, namespace, event)?;
                     res.history = Some(history);
                 } else {
-                    return Err(ffail!("Unexpected XML event: {event:?}"));
+                    return Err(anyhow!("Unexpected XML event: {event:?}"));
                 }
             }
 

crates/proto-xmpp/src/prelude.rs

@@ -0,0 +1,3 @@
+use std::sync::Arc;
+
+pub type Str = Arc<str>;

crates/proto-xmpp/src/roster.rs

@@ -1,7 +1,7 @@
 use quick_xml::events::{BytesStart, Event};
 
-use crate::prelude::*;
+use crate::xml::*;
-use crate::util::xml::*;
+use anyhow::{anyhow as ffail, Result};
 
 pub const XMLNS: &'static str = "jabber:iq:roster";
 

crates/proto-xmpp/src/sasl.rs

@@ -5,13 +5,15 @@ use quick_xml::{
     NsReader, Writer,
 };
 use tokio::io::{AsyncBufRead, AsyncWrite};
+use base64::{Engine as _, engine::general_purpose};
 
 use super::skip_text;
-use crate::prelude::*;
+use anyhow::{anyhow, Result};
 
 pub enum Mechanism {
     Plain,
 }
+
 impl Mechanism {
     pub fn to_str(&self) -> &'static str {
         match self {
@@ -22,15 +24,33 @@ impl Mechanism {
     pub fn from_str(input: &[u8]) -> Result<Mechanism> {
         match input {
             b"PLAIN" => Ok(Mechanism::Plain),
-            _ => Err(fail(format!("unknown auth mechanism: {input:?}").as_str())),
+            _ => Err(anyhow!("unknown auth mechanism: {input:?}")),
         }
     }
 }
 
+pub struct AuthBody {
+    pub login: String,
+    pub password: String,
+}
+
+impl AuthBody {
+    pub fn from_str(input: Vec<u8>) -> Result<AuthBody> {
+        let decoded_body = general_purpose::STANDARD.decode(input).unwrap();
+        let parsed_to_string = String::from_utf8(decoded_body).unwrap();
+        let separated_words: Vec<&str> = parsed_to_string.split("\x00").filter(|&part| !part.is_empty()).collect::<Vec<_>>().clone();
+        if &separated_words.len() == &2_usize {
+            Ok(AuthBody { login: separated_words[0].to_string(), password: separated_words[1].to_string() })
+        } else { Err(anyhow!("Incorrect auth format")) }
+    }
+}
+
+
 pub struct Auth {
     pub mechanism: Mechanism,
     pub body: Vec<u8>,
 }
+
 impl Auth {
     pub async fn parse(
         reader: &mut NsReader<impl AsyncBufRead + Unpin>,
@@ -48,20 +68,20 @@ impl Auth {
             if let Some(mechanism) = mechanism {
                 Mechanism::from_str(mechanism.borrow())?
             } else {
-                return Err(fail("expected mechanism attribute in <auth>"));
+                return Err(anyhow!("expected mechanism attribute in <auth>"));
             }
         } else {
-            return Err(fail("expected start of <auth>"));
+            return Err(anyhow!("expected start of <auth>"));
         };
         let body = if let Event::Text(text) = reader.read_event_into_async(buf).await? {
             text.into_inner().into_owned()
         } else {
-            return Err(fail("expected text body in <auth>"));
+            return Err(anyhow!("expected text body in <auth>"));
         };
         if let Event::End(_) = reader.read_event_into_async(buf).await? {
             //TODO
         } else {
-            return Err(fail("expected end of <auth>"));
+            return Err(anyhow!("expected end of <auth>"));
         };
 
         Ok(Auth { mechanism, body })
@@ -69,6 +89,7 @@ impl Auth {
 }
 
 pub struct Success;
+
 impl Success {
     pub async fn write_xml(&self, writer: &mut Writer<impl AsyncWrite + Unpin>) -> Result<()> {
         let event = BytesStart::new(r#"success xmlns="urn:ietf:params:xml:ns:xmpp-sasl""#);

crates/proto-xmpp/src/session.rs

@@ -1,7 +1,7 @@
 use quick_xml::events::{BytesStart, Event};
 
-use crate::prelude::*;
+use crate::xml::*;
-use crate::util::xml::*;
+use anyhow::{anyhow, Result};
 
 pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-session";
 
@@ -29,11 +29,11 @@ impl Parser for SessionParser {
                     Continuation::Continue(SessionParser(SessionParserInner::InSession))
                 }
                 Event::Empty(_) => Continuation::Final(Ok(Session)),
-                _ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))),
+                _ => Continuation::Final(Err(anyhow!("Unexpected XML event: {event:?}"))),
             },
             SessionParserInner::InSession => match event {
                 Event::End(_) => Continuation::Final(Ok(Session)),
-                _ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))),
+                _ => Continuation::Final(Err(anyhow!("Unexpected XML event: {event:?}"))),
             },
         }
     }

crates/proto-xmpp/src/stream.rs

@@ -5,8 +5,9 @@ use quick_xml::{NsReader, Writer};
 use tokio::io::{AsyncBufRead, AsyncWrite};
 
 use super::skip_text;
-use crate::prelude::*;
+
-use crate::util::xml::ToXml;
+use anyhow::{anyhow, Result};
+use crate::xml::ToXml;
 
 pub static XMLNS: &'static str = "http://etherx.jabber.org/streams";
 pub static PREFIX: &'static str = "stream";
@@ -63,7 +64,6 @@ impl ClientStreamStart {
                 version: version.unwrap(),
             })
         } else {
-            log::error!("WAT: {incoming:?}");
             Err(panic!())
         }
     }

crates/proto-xmpp/src/tls.rs

@@ -5,7 +5,8 @@ use quick_xml::{NsReader, Writer};
 use tokio::io::{AsyncBufRead, AsyncWrite};
 
 use super::skip_text;
-use crate::prelude::*;
+
+use anyhow::{anyhow, Result};
 
 pub static XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-tls";
 
@@ -21,7 +22,7 @@ impl StartTLS {
                 return Ok(StartTLS);
             }
         }
-        Err(ffail!("XML tag starttls expected, received: {incoming:?}"))
+        Err(anyhow!("XML tag starttls expected, received: {incoming:?}"))
     }
 }
 

crates/proto-xmpp/src/xml/mod.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use quick_xml::events::Event;
 use quick_xml::name::ResolveResult;
 
-use crate::prelude::Result;
+use anyhow::{anyhow, Result};
 
 mod ignore;
 pub use ignore::Ignore;
@@ -72,6 +72,7 @@ macro_rules! fail_fast {
     };
 }
 
+#[macro_export]
 macro_rules! delegate_parsing {
     ($parser: ty, $namespace: expr, $event: expr) => {{
         let mut parser = <$parser as FromXml>::parse().consume($namespace, $event);
@@ -88,6 +89,7 @@ macro_rules! delegate_parsing {
     }};
 }
 
+#[macro_export]
 macro_rules! match_parser {
     ($name: expr, $ns: expr, $event: expr; $subtype: ty, $fin: block) => {
         if $name.0 == <$subtype as FromXmlTag>::NAME.as_bytes() && $ns == ResolveResult::Bound(Namespace(<$subtype as FromXmlTag>::NS.as_bytes())) {
@@ -105,6 +107,6 @@ macro_rules! match_parser {
     };
 }
 
-pub(crate) use delegate_parsing;
+pub use delegate_parsing;
 pub(crate) use fail_fast;
-pub(crate) use match_parser;
+pub use match_parser;

src/http.rs

@@ -11,10 +11,10 @@ use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
 
-use crate::core::repo::Storage;
+use lavina_core::prelude::*;
-use crate::core::room::RoomRegistry;
+use lavina_core::repo::Storage;
-use crate::prelude::*;
+use lavina_core::room::RoomRegistry;
-use crate::util::Terminator;
+use lavina_core::terminator::Terminator;
 
 use mgmt_api::*;
 
@@ -31,7 +31,7 @@ pub async fn launch(
     rooms: RoomRegistry,
     storage: Storage,
 ) -> Result<Terminator> {
-    log::info!("Starting the telemetry service");
+    log::info!("Starting the http service");
     let listener = TcpListener::bind(config.listen_on).await?;
     log::debug!("Listener started");
     let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, storage, rx.map(|_| ())));
@@ -67,7 +67,7 @@ async fn main_loop(
             },
         }
     }
-    log::info!("Terminating the telemetry service");
+    log::info!("Terminating the http service");
     Ok(())
 }
 

src/main.rs

@@ -1,15 +1,4 @@
-#![feature(
+mod http;
-    generators,
-    generator_trait,
-    type_alias_impl_trait,
-    impl_trait_in_assoc_type
-)]
-
-mod core;
-mod prelude;
-mod projections;
-mod protos;
-mod util;
 
 use std::future::Future;
 use std::path::Path;
@@ -20,23 +9,23 @@ use figment::{providers::Toml, Figment};
 use prometheus::Registry as MetricsRegistry;
 use serde::Deserialize;
 
-use crate::core::player::PlayerRegistry;
+use lavina_core::player::PlayerRegistry;
-use crate::core::repo::Storage;
+use lavina_core::prelude::*;
-use crate::core::room::RoomRegistry;
+use lavina_core::repo::Storage;
-use crate::prelude::*;
+use lavina_core::room::RoomRegistry;
 
 #[derive(Deserialize, Debug)]
 struct ServerConfig {
-    telemetry: util::telemetry::ServerConfig,
+    telemetry: http::ServerConfig,
-    irc: projections::irc::ServerConfig,
+    irc: projection_irc::ServerConfig,
-    xmpp: projections::xmpp::ServerConfig,
+    xmpp: projection_xmpp::ServerConfig,
-    storage: core::repo::StorageConfig,
+    storage: lavina_core::repo::StorageConfig,
 }
 
 #[derive(Parser)]
 struct CliArgs {
     #[arg(long)]
-    config: Box<Path>, 
+    config: Box<Path>,
 }
 
 fn load_config() -> Result<ServerConfig> {
@@ -64,10 +53,16 @@ async fn main() -> Result<()> {
     let storage = Storage::open(storage_config).await?;
     let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
     let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?;
-    let telemetry_terminator =
+    let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
-        util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
+    let irc = projection_irc::launch(
-    let irc = projections::irc::launch(irc_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?;
+        irc_config,
-    let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
+        players.clone(),
+        rooms.clone(),
+        metrics.clone(),
+        storage.clone(),
+    )
+    .await?;
+    let xmpp = projection_xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?;
     tracing::info!("Started");
 
     sleep.await;

src/projections/irc/test.rs

@@ -1,166 +0,0 @@
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::tcp::{ReadHalf, WriteHalf};
-use tokio::net::TcpStream;
-
-use crate::prelude::*;
-
-struct TestScope<'a> {
-    reader: BufReader<ReadHalf<'a>>,
-    writer: WriteHalf<'a>,
-    buffer: Vec<u8>,
-}
-impl<'a> TestScope<'a> {
-    async fn send(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
-        self.writer.write_all(str.as_ref().as_bytes()).await?;
-        self.writer.flush().await?;
-        Ok(())
-    }
-
-    async fn expect(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
-        let len = self.reader.read_until(b'\n', &mut self.buffer).await?;
-        assert_eq!(std::str::from_utf8(&self.buffer[0..len])?, str.as_ref());
-        self.buffer.clear();
-        Ok(())
-    }
-
-    async fn assert(&mut self, f: impl FnOnce(&str) -> Result<()>) -> Result<()> {
-        let len = self.reader.read_until(b'\n', &mut self.buffer).await?;
-        let res = f(std::str::from_utf8(&self.buffer[0..len])?);
-        self.buffer.clear();
-        res
-    }
-}
-
-async fn init_client(stream: &mut TcpStream) -> Result<TestScope> {
-    let (reader, writer) = stream.split();
-    let reader = BufReader::new(reader);
-    let buffer = vec![];
-    Ok(TestScope {
-        reader,
-        writer,
-        buffer,
-    })
-}
-
-macro_rules! send {
-    ($scope: expr, $($arg:tt),*) => {{
-        $scope.send(&format!($($arg,)*)).await?;
-    }};
-}
-
-macro_rules! expect {
-    ($scope: expr, $($arg:tt),*) => {{
-        $scope.expect(&format!($($arg,)*)).await?;
-    }};
-}
-
-const SERVER_HOSTNAME: &'static str = "irc.localhost";
-
-#[rustfmt::skip]
-async fn registration(scope: &mut TestScope<'_>, nickname: &str) -> Result<()> {
-    expect!(scope, ":irc.localhost NOTICE * :Welcome to my server!\r\n");
-    send!(scope, "NICK {nickname}\r\n");
-    send!(scope, "USER UserName 0 * :Real Name\r\n");
-    expect!(scope, ":irc.localhost 001 {nickname} :Welcome to Kek Server\r\n");
-    expect!(scope, ":irc.localhost 002 {nickname} :Welcome to Kek Server\r\n");
-    expect!(scope, ":irc.localhost 003 {nickname} :Welcome to Kek Server\r\n");
-    expect!(scope, ":irc.localhost 004 {nickname} irc.localhost kek-0.1.alpha.3 r CFILPQbcefgijklmnopqrstvz\r\n");
-    expect!(scope, ":irc.localhost 005 {nickname} CHANTYPES=# :are supported by this server\r\n");
-    Ok(())
-}
-
-#[rustfmt::skip]
-async fn join(scope: &mut TestScope<'_>, nickname: &str, chan: &str) -> Result<()> {
-    send!(scope, "JOIN #{chan}\r\n");
-    expect!(scope, ":{nickname} JOIN #{chan}\r\n");
-    expect!(scope, ":irc.localhost 332 {nickname} #{chan} :New room\r\n");
-    expect!(scope, ":irc.localhost 353 {nickname} = #{chan} :{nickname}\r\n");
-    expect!(scope, ":irc.localhost 366 {nickname} #{chan} :End of /NAMES list\r\n");
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_registration() -> Result<()> {
-    let mut stream = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope = init_client(&mut stream).await?;
-
-    registration(&mut scope, "NickName1").await?;
-    join(&mut scope, "NickName1", "chan1").await?;
-    Ok(())
-}
-
-#[rustfmt::skip]
-#[tokio::test]
-async fn test_two_connections_one_player() -> Result<()> {
-    let mut stream1 = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope1 = init_client(&mut stream1).await?;
-    let mut stream2 = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope2 = init_client(&mut stream2).await?;
-
-    let nickname = "NickName2";
-    let chan = "chan2";
-
-    registration(&mut scope1, nickname).await?;
-    registration(&mut scope2, nickname).await?;
-    join(&mut scope1, nickname, chan).await?;
-
-    // force join on second connection when the other one joins a room
-    expect!(scope2, ":{nickname} JOIN #{chan}\r\n");
-    expect!(scope2, ":{SERVER_HOSTNAME} 332 {nickname} #{chan} :New room\r\n");
-    expect!(scope2, ":{SERVER_HOSTNAME} 353 {nickname} = #{chan} :{nickname}\r\n");
-    expect!(scope2, ":{SERVER_HOSTNAME} 366 {nickname} #{chan} :End of /NAMES list\r\n");
-
-    // force send PRIVMSG to other connections
-    send!(scope1, "PRIVMSG #{chan} :Chmoki vsem v etam chati!\r\n");
-    expect!(scope2, ":{nickname} PRIVMSG #{chan} :Chmoki vsem v etam chati!\r\n");
-    send!(scope2, "PRIVMSG #{chan} :I tebe privetiki\r\n");
-    expect!(scope1, ":{nickname} PRIVMSG #{chan} :I tebe privetiki\r\n");
-
-    let mut stream3 = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope3 = init_client(&mut stream3).await?;
-    registration(&mut scope3, nickname).await?;
-
-    // force join on registration
-    expect!(scope3, ":{nickname} JOIN #{chan}\r\n");
-    expect!(scope3, ":{SERVER_HOSTNAME} 332 {nickname} #{chan} :New room\r\n");
-    expect!(scope3, ":{SERVER_HOSTNAME} 353 {nickname} = #{chan} :{nickname}\r\n");
-    expect!(scope3, ":{SERVER_HOSTNAME} 366 {nickname} #{chan} :End of /NAMES list\r\n");
-
-    Ok(())
-}
-
-#[rustfmt::skip]
-#[tokio::test]
-async fn test_two_players() -> Result<()> {
-    let mut stream1 = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope1 = init_client(&mut stream1).await?;
-    let mut stream2 = TcpStream::connect("127.0.0.1:6667").await?;
-    let mut scope2 = init_client(&mut stream2).await?;
-
-    let nickname1 = "NickName3";
-    let nickname2 = "NickName4";
-
-    let chan = "chan3";
-
-    registration(&mut scope1, "NickName3").await?;
-    registration(&mut scope2, "NickName4").await?;
-    join(&mut scope1, nickname1, "chan3").await?;
-    send!(scope2, "JOIN #{chan}\r\n");
-    expect!(scope2, ":{nickname2} JOIN #{chan}\r\n");
-    expect!(scope2, ":irc.localhost 332 {nickname2} #{chan} :New room\r\n");
-    scope2.assert(|line| {
-        if line == format!(":irc.localhost 353 {nickname2} = #{chan} :{nickname1} {nickname2}\r\n") { return Ok(()) }
-        if line == format!(":irc.localhost 353 {nickname2} = #{chan} :{nickname2} {nickname1}\r\n") { return Ok(()) }
-        panic!("incorrect chan member list received: {line}");
-    }).await?;
-    expect!(scope2, ":irc.localhost 366 {nickname2} #{chan} :End of /NAMES list\r\n");
-
-    expect!(scope1, ":{nickname2} JOIN #{chan}\r\n");
-
-    send!(scope1, "PRIVMSG #chan3 :Chmoki vsem v etam chati!\r\n");
-    expect!(scope2, ":{nickname1} PRIVMSG #chan3 :Chmoki vsem v etam chati!\r\n");
-    send!(scope2, "PRIVMSG #chan3 :I tebe privetiki\r\n");
-    expect!(scope1, ":{nickname2} PRIVMSG #chan3 :I tebe privetiki\r\n");
-
-    Ok(())
-}

src/projections/mod.rs

@@ -1,3 +0,0 @@
-//! Protocol projections — implementations of public APIs.
-pub mod irc;
-pub mod xmpp;

src/protos/mod.rs

@@ -1,2 +0,0 @@
-//! Definitions of wire protocols to be used in implementations of projections.
-pub mod xmpp;

src/util/testkit.rs

@@ -1,16 +0,0 @@
-use std::task::{Context, Poll};
-
-use futures_util::task::noop_waker_ref;
-
-use crate::prelude::*;
-
-pub fn sync_future<T>(future: impl Future<Output = T>) -> Result<T> {
-    let waker = noop_waker_ref();
-    let mut context = Context::from_waker(waker);
-    pin!(future);
-    if let Poll::Ready(a) = future.poll(&mut context) {
-        Ok(a)
-    } else {
-        Err(anyhow::Error::msg("Future has suspended"))
-    }
-}

test/.gitignore

@@ -1 +0,0 @@
-*.log

test/init_state.sql

@@ -1,6 +0,0 @@
-insert into users(name)
-values ('kek'), ('shrek')
-returning id;
-
-insert into challenges_plain_password(user_id, password)
-values (1, 'parolchik1'), (2, 'qwerty123');

test/run.sh

@@ -1,5 +0,0 @@
-#!/bin/sh
-cargo build
-./target/debug/lavina 1>test/lavina.stdout.log 2>test/lavina.stderr.log &
-cargo test
-kill %1