.gitea/workflows/test-pr.yml

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

Cargo.lock

@@ -854,31 +854,23 @@ dependencies = [
  "futures-util",
  "http-body-util",
  "hyper 1.0.0-rc.3",
- "lavina-core",
  "mgmt-api",
  "nonempty",
- "projection-irc",
- "projection-xmpp",
  "prometheus",
+ "proto-irc",
+ "proto-xmpp",
+ "quick-xml",
  "regex",
  "reqwest",
+ "rustls-pemfile",
  "serde",
  "serde_json",
- "tokio",
- "tracing",
- "tracing-subscriber",
-]
-
-[[package]]
-name = "lavina-core"
-version = "0.0.1-dev"
-dependencies = [
- "anyhow",
- "prometheus",
- "serde",
  "sqlx",
  "tokio",
+ "tokio-rustls",
  "tracing",
+ "tracing-subscriber",
+ "uuid",
 ]
 
 [[package]]
@@ -1248,40 +1240,6 @@ 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"

Cargo.toml

@@ -1,9 +1,7 @@
 [workspace]
 members = [
     ".",
-    "crates/lavina-core",
     "crates/proto-irc",
-    "crates/projection-irc",
     "crates/proto-xmpp",
     "crates/mgmt-api",
 ]
@@ -24,9 +22,6 @@ 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 }
-lavina-core = { path = "crates/lavina-core" }
 
 [package]
 name = "lavina"
@@ -42,15 +37,19 @@ http-body-util = "0.1.0-rc.3"
 serde.workspace = true
 serde_json = "1.0.93"
 tokio.workspace = true
-tracing.workspace = true
+tracing = "0.1.37" # logging & tracing api
 tracing-subscriber = "0.3.16"
 futures-util.workspace = true
-prometheus.workspace = true
+prometheus = { version = "0.13.3", default-features = false }
 nonempty.workspace = true
+tokio-rustls = "0.24.1"
+rustls-pemfile = "1.0.2"
+quick-xml.workspace = true
 derive_more.workspace = true
-lavina-core.workspace = true
+uuid = { version = "1.3.0", features = ["v4"] }
-projection-irc = { path = "crates/projection-irc" }
+sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "migrate"] }
-projection-xmpp = { path = "crates/projection-xmpp" }
+proto-irc = { path = "crates/proto-irc" }
+proto-xmpp = { path = "crates/proto-xmpp" }
 mgmt-api = { path = "crates/mgmt-api" }
 clap.workspace = true
 

crates/lavina-core/Cargo.toml

@@ -1,12 +0,0 @@
-[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/projection-irc/Cargo.toml

@@ -1,16 +0,0 @@
-[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-xmpp/Cargo.toml

@@ -1,20 +0,0 @@
-[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/proto-xmpp/src/client.rs

@@ -3,7 +3,8 @@ use quick_xml::events::attributes::Attribute;
 use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
 use quick_xml::name::{QName, ResolveResult};
 
-use anyhow::{anyhow as ffail, Result};
+use anyhow::{Result, anyhow as ffail};
+
 
 use crate::prelude::*;
 use crate::xml::*;
@@ -13,45 +14,44 @@ use super::bind::Jid;
 pub const XMLNS: &'static str = "jabber:client";
 
 #[derive(PartialEq, Eq, Debug)]
-pub struct Message<T> {
+pub struct Message {
     pub from: Option<Jid>,
     pub id: Option<String>,
     pub to: Option<Jid>,
     // default is Normal
     pub r#type: MessageType,
     pub lang: Option<Str>,
+
     pub subject: Option<Str>,
     pub body: Str,
-    pub custom: Vec<T>,
 }
 
-impl<T: FromXml> FromXmlTag for Message<T> {
+impl FromXmlTag for Message {
     const NS: &'static str = XMLNS;
     const NAME: &'static str = "message";
 }
 
-impl<T: FromXml> FromXml for Message<T> {
+impl FromXml for Message {
-    type P = impl Parser<Output = Result<Self>>;
+    type P = MessageParser;
 
     fn parse() -> Self::P {
-        MessageParser(MessageParserInner::Init)
+        MessageParserInner::Init.into()
     }
 }
 
 #[derive(From)]
-struct MessageParser<T: FromXml>(MessageParserInner<T>);
+pub struct MessageParser(MessageParserInner);
 
 #[derive(Default)]
-enum MessageParserInner<T: FromXml> {
+enum MessageParserInner {
     #[default]
     Init,
-    Outer(MessageParserState<T>),
+    Outer(MessageParserState),
-    InSubject(MessageParserState<T>),
+    InSubject(MessageParserState),
-    InBody(MessageParserState<T>),
+    InBody(MessageParserState),
-    InCustom(MessageParserState<T>, T::P),
 }
 #[derive(Default)]
-struct MessageParserState<T> {
+struct MessageParserState {
     from: Option<Jid>,
     id: Option<String>,
     to: Option<Jid>,
@@ -59,27 +59,21 @@ struct MessageParserState<T> {
     lang: Option<Str>,
     subject: Option<Str>,
     body: Option<Str>,
-    custom: Vec<T>,
 }
-impl<T: FromXml> Parser for MessageParser<T> {
+impl Parser for MessageParser {
-    type Output = Result<Message<T>>;
+    type Output = Result<Message>;
 
-    fn consume<'a>(self: Self, namespace: ResolveResult, event: &Event<'a>) -> Continuation<Self, Self::Output> {
+    fn consume<'a>(
+        self: Self,
+        namespace: ResolveResult,
+        event: &Event<'a>,
+    ) -> Continuation<Self, Self::Output> {
         // TODO validate tag name and namespace at each stage
         use MessageParserInner::*;
         match self.0 {
             Init => {
                 if let Event::Start(ref bytes) = event {
-                    let mut state: MessageParserState<T> = MessageParserState {
+                    let mut state: MessageParserState = Default::default();
-                        from: None,
-                        id: None,
-                        to: None,
-                        r#type: MessageType::Normal,
-                        lang: None,
-                        subject: None,
-                        body: None,
-                        custom: vec![],
-                    };
                     for attr in bytes.attributes() {
                         let attr = fail_fast!(attr);
                         if attr.key.0 == b"from" {
@@ -110,7 +104,7 @@ impl<T: FromXml> Parser for MessageParser<T> {
                     } else if bytes.name().0 == b"body" {
                         Continuation::Continue(InBody(state).into())
                     } else {
-                        Continuation::Continue(InCustom(state, T::parse()).into())
+                        Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}")))
                     }
                 }
                 Event::End(_) => {
@@ -123,7 +117,6 @@ impl<T: FromXml> Parser for MessageParser<T> {
                             lang: state.lang,
                             subject: state.subject,
                             body,
-                            custom: state.custom,
                         }))
                     } else {
                         Continuation::Final(Err(ffail!("Body not found")))
@@ -151,19 +144,11 @@ impl<T: FromXml> Parser for MessageParser<T> {
                 Event::End(_) => Continuation::Continue(Outer(state).into()),
                 _ => Continuation::Final(Err(ffail!("Unexpected XML event: {event:?}"))),
             },
-            InCustom(mut state, mut custom) => match custom.consume(namespace, event) {
-                Continuation::Final(Ok(e)) => {
-                    state.custom.push(e);
-                    Continuation::Continue(Outer(state).into())
-                }
-                Continuation::Final(Err(e)) => Continuation::Final(Err(e)),
-                Continuation::Continue(c) => Continuation::Continue(InCustom(state, c).into()),
-            },
         }
     }
 }
 
-impl<T: ToXml> ToXml for Message<T> {
+impl ToXml for Message {
     fn serialize(&self, events: &mut Vec<Event<'static>>) {
         let mut bytes = BytesStart::new(format!(r#"message xmlns="{}""#, XMLNS));
         if let Some(from) = &self.from {
@@ -276,7 +261,11 @@ struct IqParserState<T> {
 impl<T: FromXml> Parser for IqParser<T> {
     type Output = Result<Iq<T>>;
 
-    fn consume<'a>(self: Self, namespace: ResolveResult, event: &Event<'a>) -> Continuation<Self, Self::Output> {
+    fn consume<'a>(
+        self: Self,
+        namespace: ResolveResult,
+        event: &Event<'a>,
+    ) -> Continuation<Self, Self::Output> {
         match self.0 {
             IqParserInner::Init => {
                 if let Event::Start(ref bytes) = event {
@@ -308,7 +297,8 @@ impl<T: FromXml> Parser for IqParser<T> {
                     Continuation::Final(Err(ffail!("Expected start")))
                 }
             }
-            IqParserInner::ParsingBody(mut state, parser) => match parser.consume(namespace, event) {
+            IqParserInner::ParsingBody(mut state, parser) => {
+                match parser.consume(namespace, event) {
                     Continuation::Final(f) => {
                         let body = fail_fast!(f);
                         state.body = Some(body);
@@ -317,7 +307,8 @@ impl<T: FromXml> Parser for IqParser<T> {
                     Continuation::Continue(parser) => {
                         Continuation::Continue(IqParser(IqParserInner::ParsingBody(state, parser)))
                     }
-            },
+                }
+            }
             IqParserInner::Final(state) => {
                 if let Event::End(ref bytes) = event {
                     let id = fail_fast!(state.id.ok_or_else(|| ffail!("No id provided")));
@@ -598,16 +589,22 @@ mod tests {
 
     #[tokio::test]
     async fn parse_message() {
-        let input = r#"<message id="aacea" type="chat" to="nikita@vlnv.dev"><subject>daa</subject><body>bbb</body><unknown-stuff></unknown-stuff></message>"#;
+        let input = r#"<message id="aacea" type="chat" to="nikita@vlnv.dev"><subject>daa</subject><body>bbb</body></message>"#;
         let mut reader = NsReader::from_reader(input.as_bytes());
         let mut buf = vec![];
-        let (ns, event) = reader.read_resolved_event_into_async(&mut buf).await.unwrap();
+        let (ns, event) = reader
+            .read_resolved_event_into_async(&mut buf)
+            .await
+            .unwrap();
         let mut parser = Message::parse().consume(ns, &event);
         let result = loop {
             match parser {
                 Continuation::Final(res) => break res,
                 Continuation::Continue(next) => {
-                    let (ns, event) = reader.read_resolved_event_into_async(&mut buf).await.unwrap();
+                    let (ns, event) = reader
+                        .read_resolved_event_into_async(&mut buf)
+                        .await
+                        .unwrap();
                     parser = next.consume(ns, &event);
                 }
             }
@@ -615,7 +612,7 @@ mod tests {
         .unwrap();
         assert_eq!(
             result,
-            Message::<Ignore> {
+            Message {
                 from: None,
                 id: Some("aacea".to_string()),
                 to: Some(Jid {
@@ -627,7 +624,6 @@ mod tests {
                 lang: None,
                 subject: Some("daa".into()),
                 body: "bbb".into(),
-                custom: vec![Ignore],
             }
         )
     }
@@ -637,13 +633,19 @@ mod tests {
         let input = r#"<iq id="bind_1" type="set"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><resource>mobile</resource></bind></iq>"#;
         let mut reader = NsReader::from_reader(input.as_bytes());
         let mut buf = vec![];
-        let (ns, event) = reader.read_resolved_event_into_async(&mut buf).await.unwrap();
+        let (ns, event) = reader
+            .read_resolved_event_into_async(&mut buf)
+            .await
+            .unwrap();
         let mut parser = Iq::<BindRequest>::parse().consume(ns, &event);
         let result = loop {
             match parser {
                 Continuation::Final(res) => break res,
                 Continuation::Continue(next) => {
-                    let (ns, event) = reader.read_resolved_event_into_async(&mut buf).await.unwrap();
+                    let (ns, event) = reader
+                        .read_resolved_event_into_async(&mut buf)
+                        .await
+                        .unwrap();
                     parser = next.consume(ns, &event);
                 }
             }

src/core/mod.rs

@@ -1,8 +1,4 @@
 //! 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;

src/core/player.rs

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

src/core/repo/mod.rs

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

src/core/room.rs

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

src/main.rs

@@ -1,4 +1,14 @@
-mod http;
+#![feature(
+    generators,
+    generator_trait,
+    type_alias_impl_trait,
+    impl_trait_in_assoc_type
+)]
+
+mod core;
+mod prelude;
+mod projections;
+mod util;
 
 use std::future::Future;
 use std::path::Path;
@@ -9,17 +19,17 @@ use figment::{providers::Toml, Figment};
 use prometheus::Registry as MetricsRegistry;
 use serde::Deserialize;
 
-use lavina_core::player::PlayerRegistry;
+use crate::core::player::PlayerRegistry;
-use lavina_core::prelude::*;
+use crate::core::repo::Storage;
-use lavina_core::repo::Storage;
+use crate::core::room::RoomRegistry;
-use lavina_core::room::RoomRegistry;
+use crate::prelude::*;
 
 #[derive(Deserialize, Debug)]
 struct ServerConfig {
-    telemetry: http::ServerConfig,
+    telemetry: util::telemetry::ServerConfig,
-    irc: projection_irc::ServerConfig,
+    irc: projections::irc::ServerConfig,
-    xmpp: projection_xmpp::ServerConfig,
+    xmpp: projections::xmpp::ServerConfig,
-    storage: lavina_core::repo::StorageConfig,
+    storage: core::repo::StorageConfig,
 }
 
 #[derive(Parser)]
@@ -53,16 +63,10 @@ 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 = http::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
+    let telemetry_terminator =
-    let irc = projection_irc::launch(
+        util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
-        irc_config,
+    let irc = projections::irc::launch(irc_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?;
-        players.clone(),
+    let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
-        rooms.clone(),
-        metrics.clone(),
-        storage.clone(),
-    )
-    .await?;
-    let xmpp = projection_xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
     tracing::info!("Started");
 
     sleep.await;

src/prelude.rs

@@ -15,3 +15,11 @@ 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;

src/projections/irc/mod.rs

@@ -1,26 +1,29 @@
 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 lavina_core::player::*;
+use crate::core::player::*;
-use lavina_core::prelude::*;
+use crate::core::repo::Storage;
-use lavina_core::repo::Storage;
+use crate::core::room::{RoomId, RoomInfo, RoomRegistry};
-use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
+use crate::prelude::*;
-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 {
@@ -67,12 +70,14 @@ async fn handle_socket(
     .await?;
     writer.flush().await?;
 
-    let registered_user: Result<RegisteredUser> = handle_registration(&mut reader, &mut writer, &mut storage).await;
+    let registered_user: Result<RegisteredUser> =
+        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).await?;
+            handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user)
+                .await?;
         }
         Err(_) => {
             log::debug!("Registration failed");
@@ -154,22 +159,23 @@ 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(anyhow!("no user found"));
+            return Err(fail("no user found"));
         }
     };
     if stored_user.password.is_none() {
         log::info!("Password not defined for user '{}'", user.nickname);
-        return Err(anyhow!("password is not defined"));
+        return Err(fail("password is not defined"));
     }
     if stored_user.password.as_deref() != pass.as_deref() {
         log::info!("Incorrect password supplied for user '{}'", user.nickname);
-        return Err(anyhow!("passwords do not match"));
+        return Err(fail("passwords do not match"));
     }
     // TODO properly implement session temination
 
@@ -244,7 +250,14 @@ async fn handle_registered_socket<'a>(
 
     let rooms_list = connection.get_rooms().await?;
     for room in &rooms_list {
-        produce_on_join_cmd_messages(&config, &user, &Chan::Global(room.id.as_inner().clone()), room, writer).await?;
+        produce_on_join_cmd_messages(
+            &config,
+            &user,
+            &Chan::Global(room.id.as_inner().clone()),
+            room,
+            writer,
+        )
+        .await?;
     }
 
     writer.flush().await?;
@@ -301,7 +314,10 @@ async fn handle_update(
 ) -> Result<()> {
     log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
     match update {
-        Updates::RoomJoined { new_member_id, room_id } => {
+        Updates::RoomJoined {
+            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;
@@ -422,14 +438,16 @@ 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.change_topic(room_id.clone(), topic.clone()).await?;
+                        user_handle
+                            .change_topic(room_id.clone(), topic.clone())
+                            .await?;
                         ServerMessage {
                             tags: vec![],
                             sender: Some(config.server_name.clone()),
@@ -551,7 +569,11 @@ async fn handle_incoming_message(
     Ok(HandleResult::Continue)
 }
 
-fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_user_nickname: &Str) -> ServerMessageBody {
+fn user_to_who_msg(
+    config: &ServerConfig,
+    requestor: &RegisteredUser,
+    target_user_nickname: &Str,
+) -> ServerMessageBody {
     // Username is equal to nickname
     let username = format!("~{target_user_nickname}").into();
 
@@ -652,13 +674,8 @@ async fn produce_on_join_cmd_messages(
     }
     .write_async(writer)
     .await?;
-    let prefixed_members: Vec<PrefixedNick> = room_info
+    let prefixed_members: Vec<PrefixedNick> = room_info.members.iter().map(|member| PrefixedNick::from_str(member.clone().into_inner())).collect();
-        .members
+    let non_empty_members: NonEmpty<PrefixedNick> = NonEmpty::from_vec(prefixed_members).unwrap_or(nonempty![PrefixedNick::from_str(user.nickname.clone())]);
-        .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![],
@@ -693,8 +710,12 @@ pub async fn launch(
 ) -> Result<Terminator> {
     log::info!("Starting IRC projection");
     let (stopped_tx, mut stopped_rx) = channel(32);
-    let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
+    let current_connections =
-    let total_connections = IntCounter::new("irc_total_connections", "Total number of opened connections")?;
+        IntGauge::new("irc_current_connections", "Open and alive TCP 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()))?;
 
@@ -759,8 +780,7 @@ pub async fn launch(
                     log::warn!("IRC connection to {socket_addr} finished with error: {err}")
                 }
             }
-        }))
+        })).await;
-        .await;
         log::info!("Stopped IRC projection");
         Ok(())
     });

src/projections/irc/test.rs

@@ -0,0 +1,166 @@
+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

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

src/projections/xmpp/mod.rs

@@ -1,5 +1,3 @@
-#![feature(generators, generator_trait, type_alias_impl_trait, impl_trait_in_assoc_type)]
-
 mod proto;
 
 use std::collections::HashMap;
@@ -9,7 +7,6 @@ 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};
@@ -22,10 +19,9 @@ use tokio::sync::mpsc::channel;
 use tokio_rustls::rustls::{Certificate, PrivateKey};
 use tokio_rustls::TlsAcceptor;
 
-use lavina_core::player::{PlayerConnection, PlayerId, PlayerRegistry};
+use crate::core::player::{PlayerConnection, PlayerId, PlayerRegistry};
-use lavina_core::prelude::*;
+use crate::core::room::{RoomId, RoomRegistry};
-use lavina_core::room::{RoomId, RoomRegistry};
+use crate::prelude::*;
-use lavina_core::terminator::Terminator;
 use proto_xmpp::bind::{BindResponse, Jid, Name, Resource, Server};
 use proto_xmpp::client::{Iq, Message, MessageType, Presence};
 use proto_xmpp::disco::*;
@@ -33,6 +29,7 @@ use proto_xmpp::roster::RosterQuery;
 use proto_xmpp::session::Session;
 use proto_xmpp::stream::*;
 use proto_xmpp::xml::{Continuation, FromXml, Parser, ToXml};
+use crate::util::Terminator;
 
 use self::proto::{ClientPacket, IqClientBody};
 
@@ -120,15 +117,21 @@ pub async fn launch(
             }
         }
         log::info!("Stopping XMPP projection");
-        join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
+        join_all(
+            actors
+                .into_iter()
+                .map(|(socket_addr, terminator)| async move {
                     log::debug!("Stopping XMPP connection at {socket_addr}");
                     match terminator.terminate().await {
                         Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
                         Err(err) => {
-                    log::warn!("XMPP connection to {socket_addr} finished with error: {err}")
+                            log::warn!(
+                                "XMPP connection to {socket_addr} finished with error: {err}"
+                            )
                         }
                     }
-        }))
+                }),
+        )
         .await;
         log::info!("Stopped XMPP projection");
         Ok(())
@@ -169,7 +172,9 @@ async fn handle_socket(
 
     let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf).await?;
     log::debug!("User authenticated");
-    let mut connection = players.connect_to_player(authenticated.player_id.clone()).await;
+    let mut connection = players
+        .connect_to_player(authenticated.player_id.clone())
+        .await;
     socket_final(
         &mut xml_reader,
         &mut xml_writer,
@@ -327,8 +332,8 @@ async fn socket_final(
             update = user_handle.receiver.recv() => {
                 if let Some(update) = update {
                     match update {
-                        lavina_core::player::Updates::NewMessage { room_id, author_id, body } => {
+                        crate::core::player::Updates::NewMessage { room_id, author_id, body } => {
-                            Message::<()> {
+                            Message {
                                 to: Some(Jid {
                                     name: Some(authenticated.xmpp_name.clone()),
                                     server: Server("localhost".into()),
@@ -344,7 +349,6 @@ async fn socket_final(
                                 lang: None,
                                 subject: None,
                                 body: body.into(),
-                                custom: vec![],
                             }
                             .serialize(&mut events);
                         }
@@ -392,9 +396,12 @@ async fn handle_packet(
             {
                 if server.0.as_ref() == "rooms.localhost" && m.r#type == MessageType::Groupchat {
                     user_handle
-                        .send_message(RoomId::from(name.0.clone())?, m.body.clone().into())
+                        .send_message(
+                            RoomId::from(name.0.clone())?,
+                            m.body.clone().into(),
+                        )
                         .await?;
-                    Message::<()> {
+                    Message {
                         to: Some(Jid {
                             name: Some(user.xmpp_name.clone()),
                             server: Server("localhost".into()),
@@ -410,7 +417,6 @@ async fn handle_packet(
                         lang: None,
                         subject: None,
                         body: m.body.clone(),
-                        custom: vec![],
                     }
                     .serialize(output);
                     false
@@ -442,7 +448,9 @@ async fn handle_packet(
                 resource: Some(resource),
             }) = p.to
             {
-                let a = user_handle.join_room(RoomId::from(name.0.clone())?).await?;
+                let a = user_handle
+                    .join_room(RoomId::from(name.0.clone())?)
+                    .await?;
                 Presence::<()> {
                     to: Some(Jid {
                         name: Some(user.xmpp_name.clone()),
@@ -625,13 +633,13 @@ async fn read_xml_header(
             if &*encoding == b"UTF-8" {
                 Ok(())
             } else {
-                Err(anyhow!("Unsupported encoding: {encoding:?}"))
+                Err(fail(format!("Unsupported encoding: {encoding:?}").as_str()))
             }
         } else {
             // Err(fail("No XML encoding provided"))
             Ok(())
         }
     } else {
-        Err(anyhow!("Expected XML header"))
+        Err(fail("Expected XML header"))
     }
 }

src/projections/xmpp/proto.rs

@@ -1,9 +1,7 @@
-use anyhow::anyhow;
 use derive_more::From;
 use quick_xml::events::Event;
 use quick_xml::name::{Namespace, ResolveResult};
 
-use lavina_core::prelude::*;
 use proto_xmpp::bind::BindRequest;
 use proto_xmpp::client::{Iq, Message, Presence};
 use proto_xmpp::disco::{InfoQuery, ItemQuery};
@@ -11,6 +9,8 @@ use proto_xmpp::roster::RosterQuery;
 use proto_xmpp::session::Session;
 use proto_xmpp::xml::*;
 
+use crate::prelude::*;
+
 #[derive(PartialEq, Eq, Debug, From)]
 pub enum IqClientBody {
     Bind(BindRequest),
@@ -29,7 +29,7 @@ impl FromXml for IqClientBody {
             let bytes = match event {
                 Event::Start(bytes) => bytes,
                 Event::Empty(bytes) => bytes,
-                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+                _ => return Err(ffail!("Unexpected XML event: {event:?}")),
             };
             let name = bytes.name();
             match_parser!(name, namespace, event;
@@ -49,7 +49,7 @@ impl FromXml for IqClientBody {
 #[derive(PartialEq, Eq, Debug, From)]
 pub enum ClientPacket {
     Iq(Iq<IqClientBody>),
-    Message(Message<Ignore>),
+    Message(Message),
     Presence(Presence<Ignore>),
     StreamEnd,
 }
@@ -65,9 +65,9 @@ impl FromXml for ClientPacket {
                     match_parser!(name, namespace, event;
                         Iq::<IqClientBody>,
                         Presence::<Ignore>,
-                        Message::<Ignore>,
+                        Message,
                         {
-                            Err(anyhow!(
+                            Err(ffail!(
                                 "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(anyhow!("Unexpected XML event: {event:?}"));
+                        return Err(ffail!("Unexpected XML event: {event:?}"));
                     }
                 }
                 _ => {
-                    return Err(anyhow!("Unexpected XML event: {event:?}"));
+                    return Err(ffail!("Unexpected XML event: {event:?}"));
                 }
             }
         }

src/util/mod.rs

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

src/util/telemetry.rs

@@ -11,10 +11,10 @@ use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
 
-use lavina_core::prelude::*;
+use crate::core::repo::Storage;
-use lavina_core::repo::Storage;
+use crate::core::room::RoomRegistry;
-use lavina_core::room::RoomRegistry;
+use crate::prelude::*;
-use lavina_core::terminator::Terminator;
+use crate::util::Terminator;
 
 use mgmt_api::*;
 
@@ -31,7 +31,7 @@ pub async fn launch(
     rooms: RoomRegistry,
     storage: Storage,
 ) -> Result<Terminator> {
-    log::info!("Starting the http service");
+    log::info!("Starting the telemetry 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 http service");
+    log::info!("Terminating the telemetry service");
     Ok(())
 }
 

src/util/testkit.rs

@@ -0,0 +1,16 @@
+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"))
+    }
+}