forked from lavina/lavina
1
0
Fork 0

Compare commits

...

6 Commits

23 changed files with 281 additions and 458 deletions

View File

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

56
Cargo.lock generated
View File

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

View File

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

View File

@ -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

View File

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

View File

@ -19,11 +19,9 @@ use tokio::{
task::JoinHandle, 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. /// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] #[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> { 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<()> { 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>> { pub async fn get_rooms(&self) -> Result<Vec<RoomInfo>> {
let (promise, deferred) = oneshot(); let (promise, deferred) = oneshot();
self.player_handle self.player_handle.send(PlayerCommand::GetRooms(promise)).await;
.send(PlayerCommand::GetRooms(promise))
.await;
Ok(deferred.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 (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; let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
Ok(deferred.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 (promise, deferred) = oneshot();
let cmd = Cmd::JoinRoom { room_id, promise }; let cmd = Cmd::JoinRoom { room_id, promise };
let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await; let _ = self.tx.send(PlayerCommand::Cmd(cmd, connection_id)).await;
@ -230,12 +211,8 @@ pub enum Updates {
#[derive(Clone)] #[derive(Clone)]
pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>); pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry { 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()))?; metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner { let inner = PlayerRegistryInner {
room_registry, room_registry,
@ -375,8 +352,7 @@ impl Player {
return; 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()); self.my_rooms.insert(room_id.clone(), room.clone());
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
let _ = promise.send(JoinResult::Success(room_info)); let _ = promise.send(JoinResult::Success(room_info));
@ -399,15 +375,10 @@ impl Player {
}; };
self.broadcast_update(update, connection_id).await; 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; let room = self.rooms.get_room(&room_id).await;
if let Some(room) = room { 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 { } else {
tracing::info!("no room found"); tracing::info!("no room found");
} }
@ -426,8 +397,7 @@ impl Player {
} => { } => {
let room = self.rooms.get_room(&room_id).await; let room = self.rooms.get_room(&room_id).await;
if let Some(mut room) = room { 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 { } else {
tracing::info!("no room found"); tracing::info!("no room found");
} }

View File

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

View File

@ -93,10 +93,10 @@ impl Storage {
Ok(()) Ok(())
} }
pub async fn close(mut self) -> Result<()> { pub async fn close(self) -> Result<()> {
let res = match Arc::try_unwrap(self.conn) { let res = match Arc::try_unwrap(self.conn) {
Ok(e) => e, 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(); let res = res.into_inner();
res.close().await?; res.close().await?;

View File

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

View File

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

View File

@ -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" }

View File

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

View File

@ -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

View File

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

View File

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

View File

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

View File

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

View File

@ -1,14 +1,4 @@
#![feature( mod http;
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::future::Future;
use std::path::Path; use std::path::Path;
@ -19,23 +9,23 @@ use figment::{providers::Toml, Figment};
use prometheus::Registry as MetricsRegistry; use prometheus::Registry as MetricsRegistry;
use serde::Deserialize; 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)] #[derive(Deserialize, Debug)]
struct ServerConfig { 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)] #[derive(Parser)]
struct CliArgs { struct CliArgs {
#[arg(long)] #[arg(long)]
config: Box<Path>, config: Box<Path>,
} }
fn load_config() -> Result<ServerConfig> { fn load_config() -> Result<ServerConfig> {
@ -63,10 +53,16 @@ async fn main() -> Result<()> {
let storage = Storage::open(storage_config).await?; let storage = Storage::open(storage_config).await?;
let rooms = RoomRegistry::new(&mut metrics, storage.clone())?; let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?; 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()).await?;
tracing::info!("Started"); tracing::info!("Started");
sleep.await; sleep.await;

View File

@ -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(())
}

View File

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

View File

@ -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"))
}
}