nikita
/
lavina
forked from lavina/lavina
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: nikita:2694936ca5523c3c1166b5416cd44e555d5a5dde
Branches Tags
nikita:main
nikita:x-user-and-items
nikita:diagram
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2
..
compare: nikita:fd96dd7449df3a450ae60a8e4581cf3a7824f114
Branches Tags
nikita:x-user-and-items
nikita:main
nikita:diagram
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2

1 Commits
2694936ca5 ... fd96dd7449

Author SHA1 Message Date
Nikita Vilunov fd96dd7449 add async-scoped 2023-09-06 15:44:37 +02:00
8 changed files with 222 additions and 212 deletions
Show Stats Expand all files Collapse all files

90
src/core/player.rs
View File

@ -7,17 +7,23 @@
//! //!
//! A player actor is a serial handler of commands from a single player. It is preferable to run all per-player validations in the player actor, //! A player actor is a serial handler of commands from a single player. It is preferable to run all per-player validations in the player actor,
//! so that they don't overload the room actor. //! so that they don't overload the room actor.
use std::collections::{HashMap, HashSet}; use std::{
use std::sync::RwLock; collections::{HashMap, HashSet},
sync::{Arc, RwLock},
};
use futures_util::FutureExt;
use prometheus::{IntGauge, Registry as MetricsRegistry}; use prometheus::{IntGauge, Registry as MetricsRegistry};
use serde::Serialize; use serde::Serialize;
use tokio::sync::mpsc::{channel, Receiver, Sender}; use tokio::{
sync::mpsc::{channel, Receiver, Sender},
task::JoinHandle,
};
use crate::util::table::{AnonTable, Key as AnonKey}; use crate::{
use crate::prelude::*; core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry},
use crate::core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry}; prelude::*,
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)]
@ -102,7 +108,7 @@ impl PlayerConnection {
} }
/// Handle to a player actor. /// Handle to a player actor.
#[derive(Clone, Debug)] #[derive(Clone)]
pub struct PlayerHandle { pub struct PlayerHandle {
tx: Sender<PlayerCommand>, tx: Sender<PlayerCommand>,
} }
@ -148,9 +154,7 @@ impl PlayerHandle {
} }
async fn send(&self, command: PlayerCommand) { async fn send(&self, command: PlayerCommand) {
if let Err(e) = self.tx.send(command).await { let _ = self.tx.send(command).await;
log::warn!("Failed to send command to a player: {e:?}");
}
} }
pub async fn update(&self, update: Updates) { pub async fn update(&self, update: Updates) {
@ -158,7 +162,6 @@ impl PlayerHandle {
} }
} }
#[derive(Debug)]
enum PlayerCommand { enum PlayerCommand {
/** Commands from connections */ /** Commands from connections */
AddConnection { AddConnection {
@ -171,10 +174,8 @@ enum PlayerCommand {
GetRooms(Promise<Vec<RoomInfo>>), GetRooms(Promise<Vec<RoomInfo>>),
/** Events from rooms */ /** Events from rooms */
Update(Updates), Update(Updates),
Stop,
} }
#[derive(Debug)]
pub enum Cmd { pub enum Cmd {
JoinRoom { JoinRoom {
room_id: RoomId, room_id: RoomId,
@ -196,7 +197,6 @@ pub enum Cmd {
}, },
} }
#[derive(Debug)]
pub enum JoinResult { pub enum JoinResult {
Success(RoomInfo), Success(RoomInfo),
Banned, Banned,
@ -227,75 +227,74 @@ pub enum Updates {
} }
/// Handle to a player registry — a shared data structure containing information about players. /// Handle to a player registry — a shared data structure containing information about players.
pub struct PlayerRegistry<'a>(RwLock<PlayerRegistryInner<'a>>); #[derive(Clone)]
impl<'a> PlayerRegistry<'a> { pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry {
pub fn empty( pub fn empty(
room_registry: &'a RoomRegistry<'a>, room_registry: RoomRegistry,
metrics: &MetricsRegistry, metrics: &mut MetricsRegistry,
) -> Result<PlayerRegistry<'a>> { ) -> Result<PlayerRegistry> {
let metric_active_players = let metric_active_players =
IntGauge::new("chat_players_active", "Number of alive player actors")?; 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 scope = unsafe { Scope::create() };
let inner = PlayerRegistryInner { let inner = PlayerRegistryInner {
room_registry, room_registry,
players: HashMap::new(), players: HashMap::new(),
metric_active_players, metric_active_players,
scope,
}; };
Ok(PlayerRegistry(RwLock::new(inner))) Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
} }
pub async fn get_or_create_player(&self, id: PlayerId) -> PlayerHandle { pub async fn get_or_create_player(&mut self, id: PlayerId) -> PlayerHandle {
let mut inner = self.0.write().unwrap(); let mut inner = self.0.write().unwrap();
if let Some(handle) = inner.players.get(&id) { if let Some((handle, _)) = inner.players.get(&id) {
handle.clone() handle.clone()
} else { } else {
let handle = Player::launch(id.clone(), inner.room_registry, &mut inner.scope); let (handle, fiber) = Player::launch(id.clone(), inner.room_registry.clone());
inner.players.insert(id, handle.clone()); inner.players.insert(id, (handle.clone(), fiber));
inner.metric_active_players.inc(); inner.metric_active_players.inc();
handle handle
} }
} }
pub async fn connect_to_player(&self, id: PlayerId) -> PlayerConnection { pub async fn connect_to_player(&mut self, id: PlayerId) -> PlayerConnection {
let player_handle = self.get_or_create_player(id).await; let player_handle = self.get_or_create_player(id).await;
player_handle.subscribe().await player_handle.subscribe().await
} }
pub async fn shutdown_all(self) -> Result<()> { pub async fn shutdown_all(&mut self) -> Result<()> {
let mut inner = self.0.write().unwrap(); let mut inner = self.0.write().unwrap();
for (i, k) in inner.players.drain() { let mut players = HashMap::new();
k.send(PlayerCommand::Stop).await; std::mem::swap(&mut players, &mut inner.players);
for (i, (k, j)) in inner.players.drain() {
drop(k); drop(k);
log::debug!("Stopping player #{i:?}") j.await?;
log::debug!("Player stopped #{i:?}")
} }
let _ = inner.scope.collect().await;
log::debug!("All players stopped"); log::debug!("All players stopped");
Ok(()) Ok(())
} }
} }
/// The player registry state representation. /// The player registry state representation.
struct PlayerRegistryInner<'a> { struct PlayerRegistryInner {
room_registry: &'a RoomRegistry<'a>, room_registry: RoomRegistry,
players: HashMap<PlayerId, PlayerHandle>, players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
metric_active_players: IntGauge, metric_active_players: IntGauge,
scope: Scope<'a>,
} }
/// Player actor inner state representation. /// Player actor inner state representation.
struct Player<'a> { struct Player {
player_id: PlayerId, player_id: PlayerId,
connections: AnonTable<Sender<Updates>>, connections: AnonTable<Sender<Updates>>,
my_rooms: HashMap<RoomId, RoomHandle<'a>>, my_rooms: HashMap<RoomId, RoomHandle>,
banned_from: HashSet<RoomId>, banned_from: HashSet<RoomId>,
rx: Receiver<PlayerCommand>, rx: Receiver<PlayerCommand>,
handle: PlayerHandle, handle: PlayerHandle,
rooms: &'a RoomRegistry<'a>, rooms: RoomRegistry,
} }
impl<'a> Player<'a> { impl Player {
fn launch(player_id: PlayerId, rooms: &'a RoomRegistry<'a>, scope: &mut Scope<'a>) -> PlayerHandle { fn launch(player_id: PlayerId, rooms: RoomRegistry) -> (PlayerHandle, JoinHandle<Player>) {
let (tx, rx) = channel(32); let (tx, rx) = channel(32);
let handle = PlayerHandle { tx }; let handle = PlayerHandle { tx };
let handle_clone = handle.clone(); let handle_clone = handle.clone();
@ -308,11 +307,11 @@ impl<'a> Player<'a> {
handle, handle,
rooms, rooms,
}; };
scope.spawn(player.main_loop().map(|_| ())); let fiber = tokio::task::spawn(player.main_loop());
handle_clone (handle_clone, fiber)
} }
async fn main_loop(mut self) -> Player<'a> { async fn main_loop(mut self) -> Self {
while let Some(cmd) = self.rx.recv().await { while let Some(cmd) = self.rx.recv().await {
match cmd { match cmd {
PlayerCommand::AddConnection { sender, promise } => { PlayerCommand::AddConnection { sender, promise } => {
@ -349,7 +348,6 @@ impl<'a> Player<'a> {
} }
} }
PlayerCommand::Cmd(cmd, connection_id) => self.handle_cmd(cmd, connection_id).await, PlayerCommand::Cmd(cmd, connection_id) => self.handle_cmd(cmd, connection_id).await,
PlayerCommand::Stop => { break; }
} }
} }
log::debug!("Shutting down player actor #{:?}", self.player_id); log::debug!("Shutting down player actor #{:?}", self.player_id);

26
src/core/repo/mod.rs
View File

@ -1,6 +1,7 @@
//! Storage and persistence logic. //! Storage and persistence logic.
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc;
use serde::Deserialize; use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteConnectOptions;
@ -14,8 +15,9 @@ pub struct StorageConfig {
pub db_path: String, pub db_path: String,
} }
#[derive(Clone)]
pub struct Storage { pub struct Storage {
conn: Mutex<SqliteConnection>, conn: Arc<Mutex<SqliteConnection>>,
} }
impl Storage { impl Storage {
pub async fn open(config: StorageConfig) -> Result<Storage> { pub async fn open(config: StorageConfig) -> Result<Storage> {
@ -27,11 +29,11 @@ impl Storage {
migrator.run(&mut conn).await?; migrator.run(&mut conn).await?;
log::info!("Migrations passed"); log::info!("Migrations passed");
let conn = Mutex::new(conn); let conn = Arc::new(Mutex::new(conn));
Ok(Storage { conn }) Ok(Storage { conn })
} }
pub async fn retrieve_user_by_name(&self, name: &str) -> Result<Option<StoredUser>> { pub async fn retrieve_user_by_name(&mut self, name: &str) -> Result<Option<StoredUser>> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
let res = sqlx::query_as( let res = sqlx::query_as(
"select u.id, u.name, c.password "select u.id, u.name, c.password
@ -45,7 +47,7 @@ impl Storage {
Ok(res) Ok(res)
} }
pub async fn retrieve_room_by_name(&self, name: &str) -> Result<Option<StoredRoom>> { pub async fn retrieve_room_by_name(&mut self, name: &str) -> Result<Option<StoredRoom>> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
let res = sqlx::query_as( let res = sqlx::query_as(
"select id, name, topic, message_count "select id, name, topic, message_count
@ -59,7 +61,7 @@ impl Storage {
Ok(res) Ok(res)
} }
pub async fn create_new_room(&self, name: &str, topic: &str) -> Result<u32> { pub async fn create_new_room(&mut self, name: &str, topic: &str) -> Result<u32> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
let (id,): (u32,) = sqlx::query_as( let (id,): (u32,) = sqlx::query_as(
"insert into rooms(name, topic) "insert into rooms(name, topic)
@ -74,9 +76,9 @@ impl Storage {
Ok(id) Ok(id)
} }
pub async fn insert_message(&self, room_id: u32, id: u32, content: &str) -> Result<()> { pub async fn insert_message(&mut self, room_id: u32, id: u32, content: &str) -> Result<()> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
sqlx::query( let (_,): (u32,) = sqlx::query_as(
"insert into messages(room_id, id, content) "insert into messages(room_id, id, content)
values (?, ?, ?); values (?, ?, ?);
update rooms set message_count = message_count + 1 where id = ?;", update rooms set message_count = message_count + 1 where id = ?;",
@ -85,14 +87,18 @@ impl Storage {
.bind(id) .bind(id)
.bind(content) .bind(content)
.bind(room_id) .bind(room_id)
.execute(&mut *executor) .fetch_one(&mut *executor)
.await?; .await?;
Ok(()) Ok(())
} }
pub async fn close(self) -> Result<()> { pub async fn close(mut self) -> Result<()> {
let res = self.conn.into_inner(); let res = match Arc::try_unwrap(self.conn) {
Ok(e) => e,
Err(e) => return Err(fail("failed to acquire DB ownership on shutdown")),
};
let res = res.into_inner();
res.close().await?; res.close().await?;
Ok(()) Ok(())
} }

41
src/core/room.rs
View File

@ -38,9 +38,10 @@ impl RoomId {
} }
/// Shared datastructure for storing metadata about rooms. /// Shared datastructure for storing metadata about rooms.
pub struct RoomRegistry<'a>(AsyncRwLock<RoomRegistryInner<'a>>); #[derive(Clone)]
impl<'a> RoomRegistry<'a> { pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
pub fn new(metrics: &mut MetricRegistry, storage: &'a Storage) -> Result<RoomRegistry<'a>> { impl RoomRegistry {
pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> {
let metric_active_rooms = let metric_active_rooms =
IntGauge::new("chat_rooms_active", "Number of alive room actors")?; IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
metrics.register(Box::new(metric_active_rooms.clone()))?; metrics.register(Box::new(metric_active_rooms.clone()))?;
@ -49,10 +50,10 @@ impl<'a> RoomRegistry<'a> {
metric_active_rooms, metric_active_rooms,
storage, storage,
}; };
Ok(RoomRegistry(AsyncRwLock::new(inner))) Ok(RoomRegistry(Arc::new(AsyncRwLock::new(inner))))
} }
pub async fn get_or_create_room(&self, room_id: RoomId) -> Result<RoomHandle<'a>> { pub async fn get_or_create_room(&mut self, room_id: RoomId) -> Result<RoomHandle> {
let mut inner = self.0.write().await; let mut inner = self.0.write().await;
if let Some(room_handle) = inner.rooms.get(&room_id) { if let Some(room_handle) = inner.rooms.get(&room_id) {
// room was already loaded into memory // room was already loaded into memory
@ -67,7 +68,7 @@ impl<'a> RoomRegistry<'a> {
subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions
topic: stored_room.topic.into(), topic: stored_room.topic.into(),
message_count: stored_room.message_count, message_count: stored_room.message_count,
storage: inner.storage, storage: inner.storage.clone(),
}; };
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room))); let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
inner.rooms.insert(room_id, room_handle.clone()); inner.rooms.insert(room_id, room_handle.clone());
@ -84,7 +85,7 @@ impl<'a> RoomRegistry<'a> {
subscriptions: HashMap::new(), subscriptions: HashMap::new(),
topic: topic.into(), topic: topic.into(),
message_count: 0, message_count: 0,
storage: inner.storage, storage: inner.storage.clone(),
}; };
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room))); let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
inner.rooms.insert(room_id, room_handle.clone()); inner.rooms.insert(room_id, room_handle.clone());
@ -93,7 +94,7 @@ impl<'a> RoomRegistry<'a> {
} }
} }
pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle<'a>> { pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle> {
let inner = self.0.read().await; let inner = self.0.read().await;
let res = inner.rooms.get(room_id); let res = inner.rooms.get(room_id);
res.map(|r| r.clone()) res.map(|r| r.clone())
@ -113,15 +114,15 @@ impl<'a> RoomRegistry<'a> {
} }
} }
struct RoomRegistryInner<'a> { struct RoomRegistryInner {
rooms: HashMap<RoomId, RoomHandle<'a>>, rooms: HashMap<RoomId, RoomHandle>,
metric_active_rooms: IntGauge, metric_active_rooms: IntGauge,
storage: &'a Storage, storage: Storage,
} }
#[derive(Clone)] #[derive(Clone)]
pub struct RoomHandle<'a>(Arc<AsyncRwLock<Room<'a>>>); pub struct RoomHandle(Arc<AsyncRwLock<Room>>);
impl<'a> RoomHandle<'a> { impl RoomHandle {
pub async fn subscribe(&self, player_id: PlayerId, player_handle: PlayerHandle) { pub async fn subscribe(&self, player_id: PlayerId, player_handle: PlayerHandle) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
lock.add_subscriber(player_id, player_handle).await; lock.add_subscriber(player_id, player_handle).await;
@ -139,10 +140,7 @@ impl<'a> RoomHandle<'a> {
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; lock.send_message(player_id, body).await;
if let Err(err) = res {
log::warn!("Failed to send message: {err:?}");
}
} }
pub async fn get_room_info(&self) -> RoomInfo { pub async fn get_room_info(&self) -> RoomInfo {
@ -169,15 +167,15 @@ impl<'a> RoomHandle<'a> {
} }
} }
struct Room<'a> { struct Room {
storage_id: u32, storage_id: u32,
room_id: RoomId, room_id: RoomId,
subscriptions: HashMap<PlayerId, PlayerHandle>, subscriptions: HashMap<PlayerId, PlayerHandle>,
message_count: u32, message_count: u32,
topic: Str, topic: Str,
storage: &'a Storage, storage: Storage,
} }
impl<'a> Room<'a> { impl Room {
async fn add_subscriber(&mut self, player_id: PlayerId, player_handle: PlayerHandle) { async fn add_subscriber(&mut self, player_id: PlayerId, player_handle: PlayerHandle) {
tracing::info!("Adding a subscriber to room"); tracing::info!("Adding a subscriber to room");
self.subscriptions.insert(player_id.clone(), player_handle); self.subscriptions.insert(player_id.clone(), player_handle);
@ -202,7 +200,6 @@ impl<'a> Room<'a> {
} }
async fn broadcast_update(&self, update: Updates, except: &PlayerId) { async fn broadcast_update(&self, update: Updates, except: &PlayerId) {
tracing::debug!("Broadcasting an update to {} subs", self.subscriptions.len());
for (player_id, sub) in &self.subscriptions { for (player_id, sub) in &self.subscriptions {
if player_id == except { if player_id == except {
continue; continue;
@ -213,7 +210,7 @@ impl<'a> Room<'a> {
} }
} }
#[derive(Serialize, Debug)] #[derive(Serialize)]
pub struct RoomInfo { pub struct RoomInfo {
pub id: RoomId, pub id: RoomId,
pub members: Vec<PlayerId>, pub members: Vec<PlayerId>,

29
src/main.rs
View File

@ -53,26 +53,29 @@ async fn main() -> Result<()> {
} = config; } = config;
let mut metrics = MetricsRegistry::new(); let mut metrics = MetricsRegistry::new();
let storage = Storage::open(storage_config).await?; let storage = Storage::open(storage_config).await?;
let rooms = RoomRegistry::new(&mut metrics, &storage)?; let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
let players = PlayerRegistry::empty(&rooms, &metrics)?; let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?;
let telemetry_terminator =
util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone()).await?;
// unsafe: outer future is never dropped, scope is joined on `scope.collect` unsafe { async_scoped::TokioScope::scope_and_collect(|s| {
let mut scope = unsafe { Scope::create() }; s.spawn(async {
let telemetry_terminator = util::telemetry::launch(telemetry_config, &metrics, &rooms, &mut scope).await?; let irc = projections::irc::launch(&irc_config, &players, &rooms, &metrics, &storage).await?;
let irc = projections::irc::launch(&irc_config, &players, &rooms, &metrics, &storage, &mut scope).await?; });
let xmpp = projections::xmpp::launch(xmpp_config, &players, &rooms, &metrics, &mut scope).await?; }) }.await;
let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
tracing::info!("Started"); tracing::info!("Started");
sleep.await; sleep.await;
tracing::info!("Begin shutdown"); tracing::info!("Begin shutdown");
let _ = xmpp.send(()); xmpp.terminate().await?;
let _ = irc.send(()); irc.terminate().await?;
let _ = telemetry_terminator.send(()); telemetry_terminator.terminate().await?;
let _ = scope.collect().await;
drop(scope);
players.shutdown_all().await?; players.shutdown_all().await?;
drop(players);
drop(rooms);
storage.close().await?; storage.close().await?;
tracing::info!("Shutdown complete"); tracing::info!("Shutdown complete");
Ok(()) Ok(())

2
src/prelude.rs
View File

@ -23,5 +23,3 @@ macro_rules! ffail {
} }
pub(crate) use ffail; pub(crate) use ffail;
pub type Scope<'a> = async_scoped::Scope<'a, (), async_scoped::Tokio>;

92
src/projections/irc/mod.rs
View File

@ -1,7 +1,7 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::net::SocketAddr; use std::net::SocketAddr;
use futures_util::FutureExt; use futures_util::future::join_all;
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};
@ -16,6 +16,7 @@ use crate::prelude::*;
use crate::protos::irc::client::{client_message, ClientMessage}; use crate::protos::irc::client::{client_message, ClientMessage};
use crate::protos::irc::server::{AwayStatus, ServerMessage, ServerMessageBody}; use crate::protos::irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
use crate::protos::irc::{Chan, Recipient}; use crate::protos::irc::{Chan, Recipient};
use crate::util::Terminator;
#[cfg(test)] #[cfg(test)]
mod test; mod test;
@ -42,10 +43,10 @@ async fn handle_socket(
config: ServerConfig, config: ServerConfig,
mut stream: TcpStream, mut stream: TcpStream,
socket_addr: &SocketAddr, socket_addr: &SocketAddr,
players: &PlayerRegistry<'_>, players: PlayerRegistry,
rooms: &RoomRegistry<'_>, rooms: RoomRegistry,
termination: Deferred<()>, // TODO use it to stop the connection gracefully termination: Deferred<()>, // TODO use it to stop the connection gracefully
storage: &Storage, mut storage: Storage,
) -> Result<()> { ) -> Result<()> {
let (reader, writer) = stream.split(); let (reader, writer) = stream.split();
let mut reader: BufReader<ReadHalf> = BufReader::new(reader); let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
@ -65,7 +66,7 @@ async fn handle_socket(
writer.flush().await?; writer.flush().await?;
let registered_user: Result<RegisteredUser> = let registered_user: Result<RegisteredUser> =
handle_registration(&mut reader, &mut writer, &storage).await; handle_registration(&mut reader, &mut writer, &mut storage).await;
match registered_user { match registered_user {
Ok(user) => { Ok(user) => {
@ -82,7 +83,7 @@ async fn handle_socket(
async fn handle_registration<'a>( async fn handle_registration<'a>(
reader: &mut BufReader<ReadHalf<'a>>, reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>, writer: &mut BufWriter<WriteHalf<'a>>,
storage: &Storage, storage: &mut Storage,
) -> Result<RegisteredUser> { ) -> Result<RegisteredUser> {
let mut buffer = vec![]; let mut buffer = vec![];
@ -175,8 +176,8 @@ async fn handle_registration<'a>(
async fn handle_registered_socket<'a>( async fn handle_registered_socket<'a>(
config: ServerConfig, config: ServerConfig,
players: &PlayerRegistry<'_>, mut players: PlayerRegistry,
rooms: &RoomRegistry<'_>, rooms: RoomRegistry,
reader: &mut BufReader<ReadHalf<'a>>, reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>, writer: &mut BufWriter<WriteHalf<'a>>,
user: RegisteredUser, user: RegisteredUser,
@ -300,7 +301,7 @@ async fn handle_update(
user: &RegisteredUser, user: &RegisteredUser,
player_id: &PlayerId, player_id: &PlayerId,
writer: &mut (impl AsyncWrite + Unpin), writer: &mut (impl AsyncWrite + Unpin),
rooms: &RoomRegistry<'_>, rooms: &RoomRegistry,
update: Updates, update: Updates,
) -> 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:?}");
@ -397,7 +398,7 @@ async fn handle_incoming_message(
buffer: &str, buffer: &str,
config: &ServerConfig, config: &ServerConfig,
user: &RegisteredUser, user: &RegisteredUser,
rooms: &RoomRegistry<'_>, rooms: &RoomRegistry,
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
writer: &mut (impl AsyncWrite + Unpin), writer: &mut (impl AsyncWrite + Unpin),
) -> Result<HandleResult> { ) -> Result<HandleResult> {
@ -688,14 +689,13 @@ async fn produce_on_join_cmd_messages(
Ok(()) Ok(())
} }
pub async fn launch<'a>( pub async fn launch(
config: &'a ServerConfig, config: &ServerConfig,
players: &'a PlayerRegistry<'_>, players: &PlayerRegistry,
rooms: &'a RoomRegistry<'_>, rooms: &RoomRegistry,
metrics: &'a MetricsRegistry, metrics: &MetricsRegistry,
storage: &'a Storage, storage: &Storage,
scope: &mut Scope<'a>, ) -> Result<Terminator> {
) -> Result<Promise<()>> {
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 =
@ -709,12 +709,10 @@ pub async fn launch<'a>(
let listener = TcpListener::bind(config.listen_on).await?; let listener = TcpListener::bind(config.listen_on).await?;
let (signal, mut rx) = oneshot(); let terminator = Terminator::spawn(|mut rx| async move {
let future = async move {
// TODO probably should separate logic for accepting new connection and storing them // TODO probably should separate logic for accepting new connection and storing them
// into two tasks so that they don't block each other // into two tasks so that they don't block each other
let mut actors = HashMap::new(); let mut actors = HashMap::new();
let mut scope = unsafe { Scope::create() };
loop { loop {
select! { select! {
biased; biased;
@ -737,22 +735,23 @@ pub async fn launch<'a>(
continue; continue;
} }
let terminator = Terminator::spawn(|termination| {
let current_connections_clone = current_connections.clone(); let players = players.clone();
let stopped_tx = stopped_tx.clone(); let rooms = rooms.clone();
let current_connections_clone = current_connections.clone();
let (a, rx) = oneshot(); let stopped_tx = stopped_tx.clone();
let future = async move { let storage = storage.clone();
match handle_socket(config, stream, &socket_addr, players, rooms, rx, storage).await { async move {
Ok(_) => log::info!("Connection terminated"), match handle_socket(config, stream, &socket_addr, players, rooms, termination, storage).await {
Err(err) => log::warn!("Connection failed: {err}"), Ok(_) => log::info!("Connection terminated"),
Err(err) => log::warn!("Connection failed: {err}"),
}
current_connections_clone.dec();
stopped_tx.send(socket_addr).await?;
Ok(())
} }
current_connections_clone.dec(); });
stopped_tx.send(socket_addr).await?; actors.insert(socket_addr, terminator);
Ok(())
};
scope.spawn(future.map(|_: Result<()>| ()));
actors.insert(socket_addr, a);
}, },
Err(err) => log::warn!("Failed to accept new connection: {err}"), Err(err) => log::warn!("Failed to accept new connection: {err}"),
} }
@ -761,20 +760,19 @@ pub async fn launch<'a>(
} }
log::info!("Stopping IRC projection"); log::info!("Stopping IRC projection");
for (socket_addr, terminator) in actors { join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
match terminator.send(()) { log::debug!("Stopping IRC connection at {socket_addr}");
Ok(_) => log::debug!("Stopping IRC connection at {socket_addr}"), match terminator.terminate().await {
Err(_) => log::debug!("IRC connection at {socket_addr} already stopped") Ok(_) => log::debug!("Stopped IRC connection at {socket_addr}"),
Err(err) => {
log::warn!("IRC connection to {socket_addr} finished with error: {err}")
}
} }
} })).await;
let _ = scope.collect().await;
drop(scope);
log::info!("Stopped IRC projection"); log::info!("Stopped IRC projection");
Ok(()) Ok(())
}; });
scope.spawn(future.map(|_: Result<()>| ()));
log::info!("Started IRC projection"); log::info!("Started IRC projection");
Ok(signal) Ok(terminator)
} }

80
src/projections/xmpp/mod.rs
View File

@ -7,7 +7,7 @@ use std::net::SocketAddr;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use futures_util::FutureExt; 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};
use quick_xml::{NsReader, Writer}; use quick_xml::{NsReader, Writer};
@ -30,6 +30,7 @@ use crate::protos::xmpp::roster::RosterQuery;
use crate::protos::xmpp::session::Session; use crate::protos::xmpp::session::Session;
use crate::protos::xmpp::stream::*; use crate::protos::xmpp::stream::*;
use crate::util::xml::{Continuation, FromXml, Parser, ToXml}; use crate::util::xml::{Continuation, FromXml, Parser, ToXml};
use crate::util::Terminator;
use self::proto::{ClientPacket, IqClientBody}; use self::proto::{ClientPacket, IqClientBody};
@ -52,13 +53,12 @@ struct Authenticated {
xmpp_muc_name: Resource, xmpp_muc_name: Resource,
} }
pub async fn launch<'a>( pub async fn launch(
config: ServerConfig, config: ServerConfig,
players: &'a PlayerRegistry<'_>, players: PlayerRegistry,
rooms: &'a RoomRegistry<'_>, rooms: RoomRegistry,
metrics: &'a MetricsRegistry, metrics: MetricsRegistry,
scope: &mut Scope<'a>, ) -> Result<Terminator> {
) -> Result<Promise<()>> {
log::info!("Starting XMPP projection"); log::info!("Starting XMPP projection");
let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?; let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?;
@ -75,16 +75,13 @@ pub async fn launch<'a>(
}); });
let listener = TcpListener::bind(config.listen_on).await?; let listener = TcpListener::bind(config.listen_on).await?;
let terminator = Terminator::spawn(|mut termination| async move {
let (signal, mut rx) = oneshot();
let future = async move {
let (stopped_tx, mut stopped_rx) = channel(32); let (stopped_tx, mut stopped_rx) = channel(32);
let mut actors = HashMap::new(); let mut actors = HashMap::new();
let mut scope = unsafe { Scope::create() };
loop { loop {
select! { select! {
biased; biased;
_ = &mut rx => break, _ = &mut termination => break,
stopped = stopped_rx.recv() => match stopped { stopped = stopped_rx.recv() => match stopped {
Some(stopped) => { let _ = actors.remove(&stopped); }, Some(stopped) => { let _ = actors.remove(&stopped); },
None => unreachable!(), None => unreachable!(),
@ -98,20 +95,22 @@ pub async fn launch<'a>(
// TODO kill the older connection and restart it // TODO kill the older connection and restart it
continue; continue;
} }
let stopped_tx = stopped_tx.clone(); let players = players.clone();
let loaded_config = loaded_config.clone(); let rooms = rooms.clone();
let (a, rx) = oneshot(); let terminator = Terminator::spawn(|termination| {
let future = async move { let stopped_tx = stopped_tx.clone();
match handle_socket(loaded_config, stream, &socket_addr, players, rooms, rx).await { let loaded_config = loaded_config.clone();
async move {
match handle_socket(loaded_config, stream, &socket_addr, players, rooms, termination).await {
Ok(_) => log::info!("Connection terminated"), Ok(_) => log::info!("Connection terminated"),
Err(err) => log::warn!("Connection failed: {err}"), Err(err) => log::warn!("Connection failed: {err}"),
} }
stopped_tx.send(socket_addr).await?; stopped_tx.send(socket_addr).await?;
Ok(()) Ok(())
}; }
scope.spawn(future.map(|_: Result<()>| ())); });
actors.insert(socket_addr, a); actors.insert(socket_addr, terminator);
}, },
Err(err) => log::warn!("Failed to accept new connection: {err}"), Err(err) => log::warn!("Failed to accept new connection: {err}"),
} }
@ -119,28 +118,35 @@ pub async fn launch<'a>(
} }
} }
log::info!("Stopping XMPP projection"); log::info!("Stopping XMPP projection");
for (socket_addr, terminator) in actors { join_all(
match terminator.send(()) { actors
Ok(_) => log::debug!("Stopping XMPP connection at {socket_addr}"), .into_iter()
Err(_) => log::debug!("XMPP connection at {socket_addr} already stopped") .map(|(socket_addr, terminator)| async move {
} log::debug!("Stopping XMPP connection at {socket_addr}");
} match terminator.terminate().await {
let _ = scope.collect().await; Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
drop(scope); Err(err) => {
log::warn!(
"XMPP connection to {socket_addr} finished with error: {err}"
)
}
}
}),
)
.await;
log::info!("Stopped XMPP projection"); log::info!("Stopped XMPP projection");
Ok(()) Ok(())
}; });
scope.spawn(future.map(|_: Result<()>| ()));
log::info!("Started XMPP projection"); log::info!("Started XMPP projection");
Ok(signal) Ok(terminator)
} }
async fn handle_socket( async fn handle_socket(
config: Arc<LoadedConfig>, config: Arc<LoadedConfig>,
mut stream: TcpStream, mut stream: TcpStream,
socket_addr: &SocketAddr, socket_addr: &SocketAddr,
players: &PlayerRegistry<'_>, mut players: PlayerRegistry,
rooms: &RoomRegistry<'_>, rooms: RoomRegistry,
termination: Deferred<()>, // TODO use it to stop the connection gracefully termination: Deferred<()>, // TODO use it to stop the connection gracefully
) -> Result<()> { ) -> Result<()> {
log::debug!("Received an XMPP connection from {socket_addr}"); log::debug!("Received an XMPP connection from {socket_addr}");
@ -265,7 +271,7 @@ async fn socket_final(
reader_buf: &mut Vec<u8>, reader_buf: &mut Vec<u8>,
authenticated: &Authenticated, authenticated: &Authenticated,
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
rooms: &RoomRegistry<'_>, rooms: &RoomRegistry,
) -> Result<()> { ) -> Result<()> {
read_xml_header(xml_reader, reader_buf).await?; read_xml_header(xml_reader, reader_buf).await?;
let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?; let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
@ -375,7 +381,7 @@ async fn handle_packet(
packet: ClientPacket, packet: ClientPacket,
user: &Authenticated, user: &Authenticated,
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
rooms: &RoomRegistry<'_>, rooms: &RoomRegistry,
) -> Result<bool> { ) -> Result<bool> {
Ok(match packet { Ok(match packet {
proto::ClientPacket::Iq(iq) => { proto::ClientPacket::Iq(iq) => {
@ -472,7 +478,7 @@ async fn handle_packet(
}) })
} }
async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms: &RoomRegistry<'_>) { async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms: &RoomRegistry) {
match iq.body { match iq.body {
proto::IqClientBody::Bind(b) => { proto::IqClientBody::Bind(b) => {
let req = Iq { let req = Iq {
@ -584,7 +590,7 @@ fn disco_info(to: Option<&str>, req: &InfoQuery) -> InfoQuery {
} }
} }
async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry<'_>) -> ItemQuery { async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry) -> ItemQuery {
let item = match to { let item = match to {
Some("localhost") => { Some("localhost") => {
vec![Item { vec![Item {

74
src/util/telemetry.rs
View File

@ -15,6 +15,7 @@ use crate::core::room::RoomRegistry;
use crate::prelude::*; use crate::prelude::*;
use crate::util::http::*; use crate::util::http::*;
use crate::util::Terminator;
type BoxBody = http_body_util::combinators::BoxBody<Bytes, Infallible>; type BoxBody = http_body_util::combinators::BoxBody<Bytes, Infallible>;
type HttpResult<T> = std::result::Result<T, Infallible>; type HttpResult<T> = std::result::Result<T, Infallible>;
@ -24,48 +25,51 @@ pub struct ServerConfig {
pub listen_on: SocketAddr, pub listen_on: SocketAddr,
} }
pub async fn launch<'a>( pub async fn launch(
config: ServerConfig, config: ServerConfig,
metrics: &'a MetricsRegistry, metrics: MetricsRegistry,
rooms: &'a RoomRegistry<'_>, rooms: RoomRegistry,
scope: &mut Scope<'a>, ) -> Result<Terminator> {
) -> Result<Promise<()>> {
log::info!("Starting the telemetry service"); log::info!("Starting the telemetry 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, rx.map(|_| ())));
Ok(terminator)
}
let (signal, mut rx) = oneshot(); async fn main_loop(
listener: TcpListener,
let future = async move { metrics: MetricsRegistry,
let mut scope = unsafe { Scope::create() }; rooms: RoomRegistry,
loop { termination: impl Future<Output = ()>,
select! { ) -> Result<()> {
biased; pin!(termination);
_ = &mut rx => break, loop {
result = listener.accept() => { select! {
let (stream, _) = result?; biased;
scope.spawn(async move { _ = &mut termination => break,
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(metrics, rooms, r))); result = listener.accept() => {
if let Err(err) = server.await { let (stream, _) = result?;
tracing::error!("Error serving connection: {:?}", err); let metrics = metrics.clone();
} let rooms = rooms.clone();
}); tokio::task::spawn(async move {
}, let registry = metrics.clone();
} let rooms = rooms.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), r)));
if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err);
}
});
},
} }
let _ = scope.collect().await; }
drop(scope); log::info!("Terminating the telemetry service");
log::info!("Terminating the telemetry service"); Ok(())
Ok(())
};
scope.spawn(future.map(|_: Result<()>| ()));
Ok(signal)
} }
async fn route( async fn route(
registry: &MetricsRegistry, registry: MetricsRegistry,
rooms: &RoomRegistry<'_>, rooms: RoomRegistry,
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
) -> std::result::Result<Response<BoxBody>, Infallible> { ) -> std::result::Result<Response<BoxBody>, Infallible> {
match (request.method(), request.uri().path()) { match (request.method(), request.uri().path()) {
@ -75,7 +79,7 @@ async fn route(
} }
} }
fn endpoint_metrics(registry: &MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> { fn endpoint_metrics(registry: MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> {
let mf = registry.gather(); let mf = registry.gather();
let mut buffer = vec![]; let mut buffer = vec![];
TextEncoder TextEncoder
@ -84,7 +88,7 @@ fn endpoint_metrics(registry: &MetricsRegistry) -> HttpResult<Response<Full<Byte
Ok(Response::new(Full::new(Bytes::from(buffer)))) Ok(Response::new(Full::new(Bytes::from(buffer))))
} }
async fn endpoint_rooms(rooms: &RoomRegistry<'_>) -> HttpResult<Response<Full<Bytes>>> { async fn endpoint_rooms(rooms: RoomRegistry) -> HttpResult<Response<Full<Bytes>>> {
let room_list = rooms.get_all_rooms().await; let room_list = rooms.get_all_rooms().await;
let mut buffer = vec![]; let mut buffer = vec![];
serde_json::to_writer(&mut buffer, &room_list).expect("unexpected fail when writing to vec"); serde_json::to_writer(&mut buffer, &room_list).expect("unexpected fail when writing to vec");