forked from lavina/lavina
Compare commits
No commits in common. "041c1b5fe2a12fc45c3123552cd0632289a24a00" and "36b0d50d5103f6bd86ef52680ce7f001ae3b1d7d" have entirely different histories.
041c1b5fe2
...
36b0d50d51
|
@ -1,21 +0,0 @@
|
||||||
repos:
|
|
||||||
- repo: https://github.com/pre-commit/pre-commit-hooks
|
|
||||||
rev: v4.5.0
|
|
||||||
hooks:
|
|
||||||
- id: check-toml
|
|
||||||
- id: end-of-file-fixer
|
|
||||||
- id: fix-byte-order-marker
|
|
||||||
- id: mixed-line-ending
|
|
||||||
- id: trailing-whitespace
|
|
||||||
|
|
||||||
- repo: local
|
|
||||||
hooks:
|
|
||||||
- id: fmt
|
|
||||||
name: fmt
|
|
||||||
description: Format
|
|
||||||
entry: cargo fmt
|
|
||||||
language: system
|
|
||||||
args:
|
|
||||||
- --all
|
|
||||||
types: [ rust ]
|
|
||||||
pass_filenames: false
|
|
|
@ -45,21 +45,6 @@ version = "0.2.16"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
|
checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "android-tzdata"
|
|
||||||
version = "0.1.1"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0"
|
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "android_system_properties"
|
|
||||||
version = "0.1.5"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
|
|
||||||
dependencies = [
|
|
||||||
"libc",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "anstream"
|
name = "anstream"
|
||||||
version = "0.6.13"
|
version = "0.6.13"
|
||||||
|
@ -231,20 +216,6 @@ version = "1.0.0"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
|
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "chrono"
|
|
||||||
version = "0.4.37"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "8a0d04d43504c61aa6c7531f1871dd0d418d91130162063b789da00fd7057a5e"
|
|
||||||
dependencies = [
|
|
||||||
"android-tzdata",
|
|
||||||
"iana-time-zone",
|
|
||||||
"js-sys",
|
|
||||||
"num-traits",
|
|
||||||
"wasm-bindgen",
|
|
||||||
"windows-targets 0.52.4",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "clap"
|
name = "clap"
|
||||||
version = "4.5.3"
|
version = "4.5.3"
|
||||||
|
@ -303,12 +274,6 @@ version = "0.4.0"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
|
checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "core-foundation-sys"
|
|
||||||
version = "0.8.6"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "cpufeatures"
|
name = "cpufeatures"
|
||||||
version = "0.2.12"
|
version = "0.2.12"
|
||||||
|
@ -764,29 +729,6 @@ dependencies = [
|
||||||
"tracing",
|
"tracing",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "iana-time-zone"
|
|
||||||
version = "0.1.60"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "e7ffbb5a1b541ea2561f8c41c087286cc091e21e556a4f09a8f6cbf17b69b141"
|
|
||||||
dependencies = [
|
|
||||||
"android_system_properties",
|
|
||||||
"core-foundation-sys",
|
|
||||||
"iana-time-zone-haiku",
|
|
||||||
"js-sys",
|
|
||||||
"wasm-bindgen",
|
|
||||||
"windows-core",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "iana-time-zone-haiku"
|
|
||||||
version = "0.1.2"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
|
|
||||||
dependencies = [
|
|
||||||
"cc",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "idna"
|
name = "idna"
|
||||||
version = "0.5.0"
|
version = "0.5.0"
|
||||||
|
@ -876,7 +818,6 @@ name = "lavina-core"
|
||||||
version = "0.0.2-dev"
|
version = "0.0.2-dev"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"chrono",
|
|
||||||
"prometheus",
|
"prometheus",
|
||||||
"serde",
|
"serde",
|
||||||
"sqlx",
|
"sqlx",
|
||||||
|
@ -2442,15 +2383,6 @@ version = "0.4.0"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "windows-core"
|
|
||||||
version = "0.52.0"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "33ab640c8d7e35bf8ba19b884ba838ceb4fba93a4e8c65a9059d08afcfc683d9"
|
|
||||||
dependencies = [
|
|
||||||
"windows-targets 0.52.4",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "windows-sys"
|
name = "windows-sys"
|
||||||
version = "0.48.0"
|
version = "0.48.0"
|
||||||
|
|
|
@ -10,4 +10,3 @@ serde.workspace = true
|
||||||
tokio.workspace = true
|
tokio.workspace = true
|
||||||
tracing.workspace = true
|
tracing.workspace = true
|
||||||
prometheus.workspace = true
|
prometheus.workspace = true
|
||||||
chrono = "0.4.37"
|
|
||||||
|
|
|
@ -1 +0,0 @@
|
||||||
alter table messages add column created_at text;
|
|
|
@ -7,16 +7,16 @@
|
||||||
//!
|
//!
|
||||||
//! 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::Arc;
|
collections::{HashMap, HashSet},
|
||||||
|
sync::{Arc, RwLock},
|
||||||
|
};
|
||||||
|
|
||||||
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};
|
||||||
use tokio::sync::RwLock;
|
|
||||||
|
|
||||||
use crate::prelude::*;
|
use crate::prelude::*;
|
||||||
use crate::repo::Storage;
|
|
||||||
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
|
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
|
||||||
use crate::table::{AnonTable, Key as AnonKey};
|
use crate::table::{AnonTable, Key as AnonKey};
|
||||||
|
|
||||||
|
@ -208,28 +208,23 @@ 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,
|
|
||||||
storage: Storage,
|
|
||||||
metrics: &mut MetricsRegistry,
|
|
||||||
) -> Result<PlayerRegistry> {
|
|
||||||
let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?;
|
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,
|
||||||
storage,
|
|
||||||
players: HashMap::new(),
|
players: HashMap::new(),
|
||||||
metric_active_players,
|
metric_active_players,
|
||||||
};
|
};
|
||||||
Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
|
Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn get_or_launch_player(&mut self, id: PlayerId) -> PlayerHandle {
|
pub async fn get_or_create_player(&mut self, id: PlayerId) -> PlayerHandle {
|
||||||
let mut inner = self.0.write().await;
|
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, fiber) = Player::launch(id.clone(), inner.room_registry.clone(), inner.storage.clone()).await;
|
let (handle, fiber) = Player::launch(id.clone(), inner.room_registry.clone());
|
||||||
inner.players.insert(id, (handle.clone(), fiber));
|
inner.players.insert(id, (handle.clone(), fiber));
|
||||||
inner.metric_active_players.inc();
|
inner.metric_active_players.inc();
|
||||||
handle
|
handle
|
||||||
|
@ -237,12 +232,12 @@ impl PlayerRegistry {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn connect_to_player(&mut self, id: PlayerId) -> PlayerConnection {
|
pub async fn connect_to_player(&mut self, id: PlayerId) -> PlayerConnection {
|
||||||
let player_handle = self.get_or_launch_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(&mut self) -> Result<()> {
|
pub async fn shutdown_all(&mut self) -> Result<()> {
|
||||||
let mut inner = self.0.write().await;
|
let mut inner = self.0.write().unwrap();
|
||||||
for (i, (k, j)) in inner.players.drain() {
|
for (i, (k, j)) in inner.players.drain() {
|
||||||
k.send(ActorCommand::Stop).await;
|
k.send(ActorCommand::Stop).await;
|
||||||
drop(k);
|
drop(k);
|
||||||
|
@ -257,8 +252,6 @@ impl PlayerRegistry {
|
||||||
/// The player registry state representation.
|
/// The player registry state representation.
|
||||||
struct PlayerRegistryInner {
|
struct PlayerRegistryInner {
|
||||||
room_registry: RoomRegistry,
|
room_registry: RoomRegistry,
|
||||||
storage: Storage,
|
|
||||||
/// Active player actors.
|
|
||||||
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
|
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
|
||||||
metric_active_players: IntGauge,
|
metric_active_players: IntGauge,
|
||||||
}
|
}
|
||||||
|
@ -266,31 +259,26 @@ struct PlayerRegistryInner {
|
||||||
/// Player actor inner state representation.
|
/// Player actor inner state representation.
|
||||||
struct Player {
|
struct Player {
|
||||||
player_id: PlayerId,
|
player_id: PlayerId,
|
||||||
storage_id: u32,
|
|
||||||
connections: AnonTable<Sender<Updates>>,
|
connections: AnonTable<Sender<Updates>>,
|
||||||
my_rooms: HashMap<RoomId, RoomHandle>,
|
my_rooms: HashMap<RoomId, RoomHandle>,
|
||||||
banned_from: HashSet<RoomId>,
|
banned_from: HashSet<RoomId>,
|
||||||
rx: Receiver<ActorCommand>,
|
rx: Receiver<ActorCommand>,
|
||||||
handle: PlayerHandle,
|
handle: PlayerHandle,
|
||||||
rooms: RoomRegistry,
|
rooms: RoomRegistry,
|
||||||
storage: Storage,
|
|
||||||
}
|
}
|
||||||
impl Player {
|
impl Player {
|
||||||
async fn launch(player_id: PlayerId, rooms: RoomRegistry, storage: Storage) -> (PlayerHandle, JoinHandle<Player>) {
|
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();
|
||||||
let storage_id = storage.retrieve_user_id_by_name(player_id.as_inner()).await.unwrap().unwrap();
|
|
||||||
let player = Player {
|
let player = Player {
|
||||||
player_id,
|
player_id,
|
||||||
storage_id,
|
|
||||||
connections: AnonTable::new(),
|
connections: AnonTable::new(),
|
||||||
my_rooms: HashMap::new(),
|
my_rooms: HashMap::new(),
|
||||||
banned_from: HashSet::new(),
|
banned_from: HashSet::from([RoomId::from("Empty").unwrap()]),
|
||||||
rx,
|
rx,
|
||||||
handle,
|
handle,
|
||||||
rooms,
|
rooms,
|
||||||
storage,
|
|
||||||
};
|
};
|
||||||
let fiber = tokio::task::spawn(player.main_loop());
|
let fiber = tokio::task::spawn(player.main_loop());
|
||||||
(handle_clone, fiber)
|
(handle_clone, fiber)
|
||||||
|
@ -384,7 +372,6 @@ impl Player {
|
||||||
todo!();
|
todo!();
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
room.add_member(self.player_id.clone(), self.storage_id).await;
|
|
||||||
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());
|
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;
|
||||||
|
@ -400,7 +387,6 @@ impl Player {
|
||||||
let room = self.my_rooms.remove(&room_id);
|
let room = self.my_rooms.remove(&room_id);
|
||||||
if let Some(room) = room {
|
if let Some(room) = room {
|
||||||
room.unsubscribe(&self.player_id).await;
|
room.unsubscribe(&self.player_id).await;
|
||||||
room.remove_member(&self.player_id).await;
|
|
||||||
}
|
}
|
||||||
let update = Updates::RoomLeft {
|
let update = Updates::RoomLeft {
|
||||||
room_id,
|
room_id,
|
||||||
|
@ -410,11 +396,12 @@ impl Player {
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) {
|
async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) {
|
||||||
let Some(room) = self.my_rooms.get(&room_id) else {
|
let room = self.rooms.get_room(&room_id).await;
|
||||||
|
if let Some(room) = room {
|
||||||
|
room.send_message(self.player_id.clone(), body.clone()).await;
|
||||||
|
} else {
|
||||||
tracing::info!("no room found");
|
tracing::info!("no room found");
|
||||||
return;
|
}
|
||||||
};
|
|
||||||
room.send_message(self.player_id.clone(), body.clone()).await;
|
|
||||||
let update = Updates::NewMessage {
|
let update = Updates::NewMessage {
|
||||||
room_id,
|
room_id,
|
||||||
author_id: self.player_id.clone(),
|
author_id: self.player_id.clone(),
|
||||||
|
@ -424,11 +411,12 @@ impl Player {
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn change_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) {
|
async fn change_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) {
|
||||||
let Some(room) = self.my_rooms.get(&room_id) else {
|
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;
|
||||||
|
} else {
|
||||||
tracing::info!("no room found");
|
tracing::info!("no room found");
|
||||||
return;
|
}
|
||||||
};
|
|
||||||
room.set_topic(self.player_id.clone(), new_topic.clone()).await;
|
|
||||||
let update = Updates::RoomTopicChanged { room_id, new_topic };
|
let update = Updates::RoomTopicChanged { room_id, new_topic };
|
||||||
self.broadcast_update(update, connection_id).await;
|
self.broadcast_update(update, connection_id).await;
|
||||||
}
|
}
|
||||||
|
|
|
@ -11,9 +11,6 @@ use tokio::sync::Mutex;
|
||||||
|
|
||||||
use crate::prelude::*;
|
use crate::prelude::*;
|
||||||
|
|
||||||
mod user;
|
|
||||||
mod room;
|
|
||||||
|
|
||||||
#[derive(Deserialize, Debug, Clone)]
|
#[derive(Deserialize, Debug, Clone)]
|
||||||
pub struct StorageConfig {
|
pub struct StorageConfig {
|
||||||
pub db_path: String,
|
pub db_path: String,
|
||||||
|
@ -90,15 +87,14 @@ impl Storage {
|
||||||
return Err(anyhow!("No such user"));
|
return Err(anyhow!("No such user"));
|
||||||
};
|
};
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
"insert into messages(room_id, id, content, author_id, created_at)
|
"insert into messages(room_id, id, content, author_id)
|
||||||
values (?, ?, ?, ?, ?);
|
values (?, ?, ?, ?);
|
||||||
update rooms set message_count = message_count + 1 where id = ?;",
|
update rooms set message_count = message_count + 1 where id = ?;",
|
||||||
)
|
)
|
||||||
.bind(room_id)
|
.bind(room_id)
|
||||||
.bind(id)
|
.bind(id)
|
||||||
.bind(content)
|
.bind(content)
|
||||||
.bind(author_id)
|
.bind(author_id)
|
||||||
.bind(chrono::Utc::now().to_string())
|
|
||||||
.bind(room_id)
|
.bind(room_id)
|
||||||
.execute(&mut *executor)
|
.execute(&mut *executor)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
|
@ -1,19 +0,0 @@
|
||||||
use anyhow::Result;
|
|
||||||
|
|
||||||
use crate::repo::Storage;
|
|
||||||
|
|
||||||
impl Storage {
|
|
||||||
pub async fn add_room_member(&mut self, room_id: &u32, player_id: &u32) -> Result<()> {
|
|
||||||
let mut executor = self.conn.lock().await;
|
|
||||||
sqlx::query(
|
|
||||||
"insert into memberships(user_id, room_id, status)
|
|
||||||
values (?, ?, 1);",
|
|
||||||
)
|
|
||||||
.bind(player_id)
|
|
||||||
.bind(room_id)
|
|
||||||
.execute(&mut *executor)
|
|
||||||
.await?;
|
|
||||||
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -1,15 +0,0 @@
|
||||||
use anyhow::Result;
|
|
||||||
|
|
||||||
use crate::repo::Storage;
|
|
||||||
|
|
||||||
impl Storage {
|
|
||||||
pub async fn retrieve_user_id_by_name(&self, name: &str) -> Result<Option<u32>> {
|
|
||||||
let mut executor = self.conn.lock().await;
|
|
||||||
let res: Option<(u32,)> = sqlx::query_as("select u.id from users u where u.name = ?;")
|
|
||||||
.bind(name)
|
|
||||||
.fetch_optional(&mut *executor)
|
|
||||||
.await?;
|
|
||||||
|
|
||||||
Ok(res.map(|(id,)| id))
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -1,5 +1,4 @@
|
||||||
//! Domain of rooms — chats with multiple participants.
|
//! Domain of rooms — chats with multiple participants.
|
||||||
use std::collections::HashSet;
|
|
||||||
use std::{collections::HashMap, hash::Hash, sync::Arc};
|
use std::{collections::HashMap, hash::Hash, sync::Arc};
|
||||||
|
|
||||||
use prometheus::{IntGauge, Registry as MetricRegistry};
|
use prometheus::{IntGauge, Registry as MetricRegistry};
|
||||||
|
@ -60,7 +59,6 @@ impl RoomRegistry {
|
||||||
storage_id: stored_room.id,
|
storage_id: stored_room.id,
|
||||||
room_id: room_id.clone(),
|
room_id: room_id.clone(),
|
||||||
subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions
|
subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions
|
||||||
members: HashSet::new(), // TODO load members from storage
|
|
||||||
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.clone(),
|
storage: inner.storage.clone(),
|
||||||
|
@ -78,7 +76,6 @@ impl RoomRegistry {
|
||||||
storage_id: id,
|
storage_id: id,
|
||||||
room_id: room_id.clone(),
|
room_id: room_id.clone(),
|
||||||
subscriptions: HashMap::new(),
|
subscriptions: HashMap::new(),
|
||||||
members: HashSet::new(),
|
|
||||||
topic: topic.into(),
|
topic: topic.into(),
|
||||||
message_count: 0,
|
message_count: 0,
|
||||||
storage: inner.storage.clone(),
|
storage: inner.storage.clone(),
|
||||||
|
@ -121,32 +118,12 @@ pub struct RoomHandle(Arc<AsyncRwLock<Room>>);
|
||||||
impl RoomHandle {
|
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;
|
||||||
tracing::info!("Adding a subscriber to a room");
|
lock.add_subscriber(player_id, player_handle).await;
|
||||||
lock.subscriptions.insert(player_id.clone(), player_handle);
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn add_member(&self, player_id: PlayerId, player_storage_id: u32) {
|
|
||||||
let mut lock = self.0.write().await;
|
|
||||||
tracing::info!("Adding a new member to a room");
|
|
||||||
let storage_id = lock.storage_id;
|
|
||||||
lock.members.insert(player_id.clone());
|
|
||||||
lock.storage.add_room_member(&storage_id, &player_storage_id).await.unwrap();
|
|
||||||
let update = Updates::RoomJoined {
|
|
||||||
room_id: lock.room_id.clone(),
|
|
||||||
new_member_id: player_id.clone(),
|
|
||||||
};
|
|
||||||
lock.broadcast_update(update, &player_id).await;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn unsubscribe(&self, player_id: &PlayerId) {
|
pub async fn unsubscribe(&self, player_id: &PlayerId) {
|
||||||
let mut lock = self.0.write().await;
|
let mut lock = self.0.write().await;
|
||||||
lock.subscriptions.remove(player_id);
|
lock.subscriptions.remove(player_id);
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn remove_member(&self, player_id: &PlayerId) {
|
|
||||||
let mut lock = self.0.write().await;
|
|
||||||
tracing::info!("Removing a member from a room");
|
|
||||||
lock.members.remove(player_id);
|
|
||||||
let update = Updates::RoomLeft {
|
let update = Updates::RoomLeft {
|
||||||
room_id: lock.room_id.clone(),
|
room_id: lock.room_id.clone(),
|
||||||
former_member_id: player_id.clone(),
|
former_member_id: player_id.clone(),
|
||||||
|
@ -171,7 +148,7 @@ impl RoomHandle {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn set_topic(&self, changer_id: PlayerId, new_topic: Str) {
|
pub async fn set_topic(&mut self, changer_id: PlayerId, new_topic: Str) {
|
||||||
let mut lock = self.0.write().await;
|
let mut lock = self.0.write().await;
|
||||||
lock.topic = new_topic.clone();
|
lock.topic = new_topic.clone();
|
||||||
let update = Updates::RoomTopicChanged {
|
let update = Updates::RoomTopicChanged {
|
||||||
|
@ -189,14 +166,22 @@ struct Room {
|
||||||
room_id: RoomId,
|
room_id: RoomId,
|
||||||
/// Player actors on the local node which are subscribed to this room's updates.
|
/// Player actors on the local node which are subscribed to this room's updates.
|
||||||
subscriptions: HashMap<PlayerId, PlayerHandle>,
|
subscriptions: HashMap<PlayerId, PlayerHandle>,
|
||||||
/// Members of the room.
|
|
||||||
members: HashSet<PlayerId>,
|
|
||||||
/// The total number of messages. Used to calculate the id of the new message.
|
/// The total number of messages. Used to calculate the id of the new message.
|
||||||
message_count: u32,
|
message_count: u32,
|
||||||
topic: Str,
|
topic: Str,
|
||||||
storage: Storage,
|
storage: Storage,
|
||||||
}
|
}
|
||||||
impl Room {
|
impl Room {
|
||||||
|
async fn add_subscriber(&mut self, player_id: PlayerId, player_handle: PlayerHandle) {
|
||||||
|
tracing::info!("Adding a subscriber to room");
|
||||||
|
self.subscriptions.insert(player_id.clone(), player_handle);
|
||||||
|
let update = Updates::RoomJoined {
|
||||||
|
room_id: self.room_id.clone(),
|
||||||
|
new_member_id: player_id.clone(),
|
||||||
|
};
|
||||||
|
self.broadcast_update(update, &player_id).await;
|
||||||
|
}
|
||||||
|
|
||||||
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, &*author_id.as_inner()).await?;
|
self.storage.insert_message(self.storage_id, self.message_count, &body, &*author_id.as_inner()).await?;
|
||||||
|
|
|
@ -111,36 +111,7 @@ impl TestServer {
|
||||||
})
|
})
|
||||||
.await?;
|
.await?;
|
||||||
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
||||||
let players = PlayerRegistry::empty(rooms.clone(), storage.clone(), &mut metrics).unwrap();
|
let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
|
||||||
let server = launch(config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await.unwrap();
|
|
||||||
Ok(TestServer {
|
|
||||||
metrics,
|
|
||||||
storage,
|
|
||||||
rooms,
|
|
||||||
players,
|
|
||||||
server,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn reboot(mut self) -> Result<TestServer> {
|
|
||||||
let config = ServerConfig {
|
|
||||||
listen_on: "127.0.0.1:0".parse().unwrap(),
|
|
||||||
server_name: "testserver".into(),
|
|
||||||
};
|
|
||||||
let TestServer {
|
|
||||||
mut metrics,
|
|
||||||
mut storage,
|
|
||||||
rooms,
|
|
||||||
mut players,
|
|
||||||
server,
|
|
||||||
} = self;
|
|
||||||
server.terminate().await?;
|
|
||||||
players.shutdown_all().await.unwrap();
|
|
||||||
drop(players);
|
|
||||||
drop(rooms);
|
|
||||||
let mut metrics = MetricsRegistry::new();
|
|
||||||
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
|
||||||
let players = PlayerRegistry::empty(rooms.clone(), storage.clone(), &mut metrics).unwrap();
|
|
||||||
let server = launch(config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await.unwrap();
|
let server = launch(config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await.unwrap();
|
||||||
Ok(TestServer {
|
Ok(TestServer {
|
||||||
metrics,
|
metrics,
|
||||||
|
@ -181,76 +152,6 @@ async fn scenario_basic() -> Result<()> {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn scenario_join_and_reboot() -> Result<()> {
|
|
||||||
let mut server = TestServer::start().await?;
|
|
||||||
|
|
||||||
// test scenario
|
|
||||||
|
|
||||||
server.storage.create_user("tester").await?;
|
|
||||||
server.storage.set_password("tester", "password").await?;
|
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
|
||||||
let mut s = TestScope::new(&mut stream);
|
|
||||||
|
|
||||||
// Open a connection and join a channel
|
|
||||||
|
|
||||||
s.send("PASS password").await?;
|
|
||||||
s.send("NICK tester").await?;
|
|
||||||
s.send("USER UserName 0 * :Real Name").await?;
|
|
||||||
s.expect_server_introduction("tester").await?;
|
|
||||||
s.expect_nothing().await?;
|
|
||||||
s.send("JOIN #test").await?;
|
|
||||||
s.expect(":tester JOIN #test").await?;
|
|
||||||
s.expect(":testserver 332 tester #test :New room").await?;
|
|
||||||
s.expect(":testserver 353 tester = #test :tester").await?;
|
|
||||||
s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
|
|
||||||
s.send("PRIVMSG #test :Hello").await?;
|
|
||||||
s.send("QUIT :Leaving").await?;
|
|
||||||
s.expect(":testserver ERROR :Leaving the server").await?;
|
|
||||||
s.expect_eof().await?;
|
|
||||||
stream.shutdown().await?;
|
|
||||||
|
|
||||||
// Open a new connection and expect to be force-joined to the channel
|
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
|
||||||
let mut s = TestScope::new(&mut stream);
|
|
||||||
|
|
||||||
async fn test(s: &mut TestScope<'_>) -> Result<()> {
|
|
||||||
s.send("PASS password").await?;
|
|
||||||
s.send("NICK tester").await?;
|
|
||||||
s.send("USER UserName 0 * :Real Name").await?;
|
|
||||||
s.expect_server_introduction("tester").await?;
|
|
||||||
s.expect(":tester JOIN #test").await?;
|
|
||||||
s.expect(":testserver 332 tester #test :New room").await?;
|
|
||||||
s.expect(":testserver 353 tester = #test :tester").await?;
|
|
||||||
s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
|
|
||||||
s.send("QUIT :Leaving").await?;
|
|
||||||
s.expect(":testserver ERROR :Leaving the server").await?;
|
|
||||||
s.expect_eof().await?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
test(&mut s).await?;
|
|
||||||
stream.shutdown().await?;
|
|
||||||
|
|
||||||
// Reboot the server
|
|
||||||
|
|
||||||
let server = server.reboot().await?;
|
|
||||||
|
|
||||||
// Open a new connection and expect to be force-joined to the channel
|
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
|
||||||
let mut s = TestScope::new(&mut stream);
|
|
||||||
|
|
||||||
test(&mut s).await?;
|
|
||||||
stream.shutdown().await?;
|
|
||||||
|
|
||||||
// wrap up
|
|
||||||
|
|
||||||
server.server.terminate().await?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn scenario_force_join_msg() -> Result<()> {
|
async fn scenario_force_join_msg() -> Result<()> {
|
||||||
let mut server = TestServer::start().await?;
|
let mut server = TestServer::start().await?;
|
||||||
|
@ -506,6 +407,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn terminate_socket_scenario() -> Result<()> {
|
async fn terminate_socket_scenario() -> Result<()> {
|
||||||
let mut server = TestServer::start().await?;
|
let mut server = TestServer::start().await?;
|
||||||
|
let address: SocketAddr = ("127.0.0.1:0".parse().unwrap());
|
||||||
|
|
||||||
// test scenario
|
// test scenario
|
||||||
|
|
||||||
|
|
|
@ -24,9 +24,9 @@ impl<'a> XmppConnection<'a> {
|
||||||
to: None,
|
to: None,
|
||||||
r#type: IqType::Result,
|
r#type: IqType::Result,
|
||||||
body: BindResponse(Jid {
|
body: BindResponse(Jid {
|
||||||
name: Some(self.user.xmpp_name.clone()),
|
name: Some(Name("darova".into())),
|
||||||
server: Server("localhost".into()),
|
server: Server("localhost".into()),
|
||||||
resource: Some(self.user.xmpp_resource.clone()),
|
resource: Some(Resource("kek".into())),
|
||||||
}),
|
}),
|
||||||
};
|
};
|
||||||
req.serialize(output);
|
req.serialize(output);
|
||||||
|
|
|
@ -52,17 +52,9 @@ struct LoadedConfig {
|
||||||
}
|
}
|
||||||
|
|
||||||
struct Authenticated {
|
struct Authenticated {
|
||||||
/// Identifier of the authenticated player.
|
|
||||||
///
|
|
||||||
/// Used when communicating with lavina-core on behalf of the player.
|
|
||||||
player_id: PlayerId,
|
player_id: PlayerId,
|
||||||
/// The user's XMPP name.
|
|
||||||
///
|
|
||||||
/// Used in `to` and `from` fields of XMPP messages.
|
|
||||||
xmpp_name: Name,
|
xmpp_name: Name,
|
||||||
/// The resource given to this user by the server.
|
|
||||||
xmpp_resource: Resource,
|
xmpp_resource: Resource,
|
||||||
/// The resource used by this user when joining MUCs.
|
|
||||||
xmpp_muc_name: Resource,
|
xmpp_muc_name: Resource,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -315,13 +307,11 @@ async fn socket_auth(
|
||||||
return Err(fail("passwords do not match"));
|
return Err(fail("passwords do not match"));
|
||||||
}
|
}
|
||||||
|
|
||||||
let name: Str = name.as_str().into();
|
|
||||||
|
|
||||||
Ok(Authenticated {
|
Ok(Authenticated {
|
||||||
player_id: PlayerId::from(name.clone())?,
|
player_id: PlayerId::from(name.as_str())?,
|
||||||
xmpp_name: Name(name.clone()),
|
xmpp_name: Name(name.to_string().into()),
|
||||||
xmpp_resource: Resource(name.clone()),
|
xmpp_resource: Resource(name.to_string().into()),
|
||||||
xmpp_muc_name: Resource(name.clone()),
|
xmpp_muc_name: Resource(name.to_string().into()),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
Err(e) => return Err(e),
|
Err(e) => return Err(e),
|
||||||
|
@ -372,7 +362,7 @@ async fn socket_final(
|
||||||
res = &mut next_xml_event => 's: {
|
res = &mut next_xml_event => 's: {
|
||||||
let (ns, event) = res?;
|
let (ns, event) = res?;
|
||||||
if let Event::Text(ref e) = event {
|
if let Event::Text(ref e) = event {
|
||||||
if e.iter().all(|x| *x == b'\n' || *x == b' ') {
|
if e.iter().all(|x| *x == 0xA) {
|
||||||
break 's true;
|
break 's true;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -20,7 +20,7 @@ use tokio_rustls::TlsConnector;
|
||||||
use lavina_core::player::PlayerRegistry;
|
use lavina_core::player::PlayerRegistry;
|
||||||
use lavina_core::repo::{Storage, StorageConfig};
|
use lavina_core::repo::{Storage, StorageConfig};
|
||||||
use lavina_core::room::RoomRegistry;
|
use lavina_core::room::RoomRegistry;
|
||||||
use projection_xmpp::{launch, RunningServer, ServerConfig};
|
use projection_xmpp::{launch, ServerConfig};
|
||||||
use proto_xmpp::xml::{Continuation, FromXml, Parser};
|
use proto_xmpp::xml::{Continuation, FromXml, Parser};
|
||||||
|
|
||||||
pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
|
pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
|
||||||
|
@ -122,49 +122,29 @@ impl ServerCertVerifier for IgnoreCertVerification {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
struct TestServer {
|
|
||||||
metrics: MetricsRegistry,
|
|
||||||
storage: Storage,
|
|
||||||
rooms: RoomRegistry,
|
|
||||||
players: PlayerRegistry,
|
|
||||||
server: RunningServer,
|
|
||||||
}
|
|
||||||
impl TestServer {
|
|
||||||
async fn start() -> Result<TestServer> {
|
|
||||||
let _ = tracing_subscriber::fmt::try_init();
|
|
||||||
let config = ServerConfig {
|
|
||||||
listen_on: "127.0.0.1:0".parse().unwrap(),
|
|
||||||
cert: "tests/certs/xmpp.pem".parse().unwrap(),
|
|
||||||
key: "tests/certs/xmpp.key".parse().unwrap(),
|
|
||||||
};
|
|
||||||
let mut metrics = MetricsRegistry::new();
|
|
||||||
let mut storage = Storage::open(StorageConfig {
|
|
||||||
db_path: ":memory:".into(),
|
|
||||||
})
|
|
||||||
.await?;
|
|
||||||
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
|
||||||
let players = PlayerRegistry::empty(rooms.clone(), storage.clone(), &mut metrics).unwrap();
|
|
||||||
let server = launch(config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await.unwrap();
|
|
||||||
Ok(TestServer {
|
|
||||||
metrics,
|
|
||||||
storage,
|
|
||||||
rooms,
|
|
||||||
players,
|
|
||||||
server,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn scenario_basic() -> Result<()> {
|
async fn scenario_basic() -> Result<()> {
|
||||||
let mut server = TestServer::start().await?;
|
tracing_subscriber::fmt::try_init();
|
||||||
|
let config = ServerConfig {
|
||||||
|
listen_on: "127.0.0.1:0".parse().unwrap(),
|
||||||
|
cert: "tests/certs/xmpp.pem".parse().unwrap(),
|
||||||
|
key: "tests/certs/xmpp.key".parse().unwrap(),
|
||||||
|
};
|
||||||
|
let mut metrics = MetricsRegistry::new();
|
||||||
|
let mut storage = Storage::open(StorageConfig {
|
||||||
|
db_path: ":memory:".into(),
|
||||||
|
})
|
||||||
|
.await?;
|
||||||
|
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
||||||
|
let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
|
||||||
|
let server = launch(config, players, rooms, metrics, storage.clone()).await.unwrap();
|
||||||
|
|
||||||
// test scenario
|
// test scenario
|
||||||
|
|
||||||
server.storage.create_user("tester").await?;
|
storage.create_user("tester").await?;
|
||||||
server.storage.set_password("tester", "password").await?;
|
storage.set_password("tester", "password").await?;
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
let mut stream = TcpStream::connect(server.addr).await?;
|
||||||
let mut s = TestScope::new(&mut stream);
|
let mut s = TestScope::new(&mut stream);
|
||||||
tracing::info!("TCP connection established");
|
tracing::info!("TCP connection established");
|
||||||
|
|
||||||
|
@ -189,7 +169,7 @@ async fn scenario_basic() -> Result<()> {
|
||||||
.with_no_client_auth(),
|
.with_no_client_auth(),
|
||||||
));
|
));
|
||||||
tracing::info!("Initiating TLS connection...");
|
tracing::info!("Initiating TLS connection...");
|
||||||
let mut stream = connector.connect(ServerName::IpAddress(server.server.addr.ip()), stream).await?;
|
let mut stream = connector.connect(ServerName::IpAddress(server.addr.ip()), stream).await?;
|
||||||
tracing::info!("TLS connection established");
|
tracing::info!("TLS connection established");
|
||||||
|
|
||||||
let mut s = TestScopeTls::new(&mut stream, buffer);
|
let mut s = TestScopeTls::new(&mut stream, buffer);
|
||||||
|
@ -203,20 +183,33 @@ async fn scenario_basic() -> Result<()> {
|
||||||
|
|
||||||
// wrap up
|
// wrap up
|
||||||
|
|
||||||
server.server.terminate().await?;
|
server.terminate().await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn scenario_basic_without_headers() -> Result<()> {
|
async fn scenario_basic_without_headers() -> Result<()> {
|
||||||
let mut server = TestServer::start().await?;
|
tracing_subscriber::fmt::try_init();
|
||||||
|
let config = ServerConfig {
|
||||||
|
listen_on: "127.0.0.1:0".parse().unwrap(),
|
||||||
|
cert: "tests/certs/xmpp.pem".parse().unwrap(),
|
||||||
|
key: "tests/certs/xmpp.key".parse().unwrap(),
|
||||||
|
};
|
||||||
|
let mut metrics = MetricsRegistry::new();
|
||||||
|
let mut storage = Storage::open(StorageConfig {
|
||||||
|
db_path: ":memory:".into(),
|
||||||
|
})
|
||||||
|
.await?;
|
||||||
|
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
||||||
|
let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
|
||||||
|
let server = launch(config, players, rooms, metrics, storage.clone()).await.unwrap();
|
||||||
|
|
||||||
// test scenario
|
// test scenario
|
||||||
|
|
||||||
server.storage.create_user("tester").await?;
|
storage.create_user("tester").await?;
|
||||||
server.storage.set_password("tester", "password").await?;
|
storage.set_password("tester", "password").await?;
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
let mut stream = TcpStream::connect(server.addr).await?;
|
||||||
let mut s = TestScope::new(&mut stream);
|
let mut s = TestScope::new(&mut stream);
|
||||||
tracing::info!("TCP connection established");
|
tracing::info!("TCP connection established");
|
||||||
|
|
||||||
|
@ -240,7 +233,7 @@ async fn scenario_basic_without_headers() -> Result<()> {
|
||||||
.with_no_client_auth(),
|
.with_no_client_auth(),
|
||||||
));
|
));
|
||||||
tracing::info!("Initiating TLS connection...");
|
tracing::info!("Initiating TLS connection...");
|
||||||
let mut stream = connector.connect(ServerName::IpAddress(server.server.addr.ip()), stream).await?;
|
let mut stream = connector.connect(ServerName::IpAddress(server.addr.ip()), stream).await?;
|
||||||
tracing::info!("TLS connection established");
|
tracing::info!("TLS connection established");
|
||||||
|
|
||||||
let mut s = TestScopeTls::new(&mut stream, buffer);
|
let mut s = TestScopeTls::new(&mut stream, buffer);
|
||||||
|
@ -253,20 +246,33 @@ async fn scenario_basic_without_headers() -> Result<()> {
|
||||||
|
|
||||||
// wrap up
|
// wrap up
|
||||||
|
|
||||||
server.server.terminate().await?;
|
server.terminate().await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn terminate_socket() -> Result<()> {
|
async fn terminate_socket() -> Result<()> {
|
||||||
let mut server = TestServer::start().await?;
|
tracing_subscriber::fmt::try_init();
|
||||||
|
let config = ServerConfig {
|
||||||
|
listen_on: "127.0.0.1:0".parse().unwrap(),
|
||||||
|
cert: "tests/certs/xmpp.pem".parse().unwrap(),
|
||||||
|
key: "tests/certs/xmpp.key".parse().unwrap(),
|
||||||
|
};
|
||||||
|
let mut metrics = MetricsRegistry::new();
|
||||||
|
let mut storage = Storage::open(StorageConfig {
|
||||||
|
db_path: ":memory:".into(),
|
||||||
|
})
|
||||||
|
.await?;
|
||||||
|
let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap();
|
||||||
|
let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap();
|
||||||
|
let server = launch(config, players, rooms, metrics, storage.clone()).await.unwrap();
|
||||||
|
let address: SocketAddr = ("127.0.0.1:0".parse().unwrap());
|
||||||
// test scenario
|
// test scenario
|
||||||
|
|
||||||
server.storage.create_user("tester").await?;
|
storage.create_user("tester").await?;
|
||||||
server.storage.set_password("tester", "password").await?;
|
storage.set_password("tester", "password").await?;
|
||||||
|
|
||||||
let mut stream = TcpStream::connect(server.server.addr).await?;
|
let mut stream = TcpStream::connect(server.addr).await?;
|
||||||
let mut s = TestScope::new(&mut stream);
|
let mut s = TestScope::new(&mut stream);
|
||||||
tracing::info!("TCP connection established");
|
tracing::info!("TCP connection established");
|
||||||
|
|
||||||
|
@ -292,10 +298,10 @@ async fn terminate_socket() -> Result<()> {
|
||||||
));
|
));
|
||||||
|
|
||||||
tracing::info!("Initiating TLS connection...");
|
tracing::info!("Initiating TLS connection...");
|
||||||
let mut stream = connector.connect(ServerName::IpAddress(server.server.addr.ip()), stream).await?;
|
let mut stream = connector.connect(ServerName::IpAddress(server.addr.ip()), stream).await?;
|
||||||
tracing::info!("TLS connection established");
|
tracing::info!("TLS connection established");
|
||||||
|
|
||||||
server.server.terminate().await?;
|
server.terminate().await?;
|
||||||
|
|
||||||
assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);
|
assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);
|
||||||
|
|
||||||
|
|
|
@ -11,15 +11,12 @@ pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
|
||||||
|
|
||||||
// TODO remove `pub` in newtypes, introduce validation
|
// TODO remove `pub` in newtypes, introduce validation
|
||||||
|
|
||||||
/// Name (node identifier) of an XMPP entity. Placed before the `@` in a JID.
|
|
||||||
#[derive(PartialEq, Eq, Debug, Clone)]
|
#[derive(PartialEq, Eq, Debug, Clone)]
|
||||||
pub struct Name(pub Str);
|
pub struct Name(pub Str);
|
||||||
|
|
||||||
/// Server name of an XMPP entity. Placed after the `@` and before the `/` in a JID.
|
|
||||||
#[derive(PartialEq, Eq, Debug, Clone)]
|
#[derive(PartialEq, Eq, Debug, Clone)]
|
||||||
pub struct Server(pub Str);
|
pub struct Server(pub Str);
|
||||||
|
|
||||||
/// Resource of an XMPP entity. Placed after the `/` in a JID.
|
|
||||||
#[derive(PartialEq, Eq, Debug, Clone)]
|
#[derive(PartialEq, Eq, Debug, Clone)]
|
||||||
pub struct Resource(pub Str);
|
pub struct Resource(pub Str);
|
||||||
|
|
||||||
|
|
|
@ -8,12 +8,11 @@ Some useful commands for development and testing.
|
||||||
|
|
||||||
Following commands require `OpenSSL` to be installed. It is provided as `openssl` package in Arch Linux.
|
Following commands require `OpenSSL` to be installed. It is provided as `openssl` package in Arch Linux.
|
||||||
|
|
||||||
Generate self-signed TLS certificate. Mind the common name (CN) field, it should match the domain name of the server.
|
Generate self-signed TLS certificate:
|
||||||
Example for localhost:
|
|
||||||
|
|
||||||
openssl req -x509 -newkey rsa:4096 -sha256 -days 365 -noenc \
|
openssl req -x509 -newkey rsa:4096 -sha256 -days 365 -noenc \
|
||||||
-keyout certs/xmpp.key -out certs/xmpp.pem \
|
-keyout certs/xmpp.key -out certs/xmpp.pem \
|
||||||
-subj "/CN=localhost"
|
-subj "/CN=example.com"
|
||||||
|
|
||||||
Print content of a TLS certificate:
|
Print content of a TLS certificate:
|
||||||
|
|
||||||
|
|
|
@ -52,7 +52,7 @@ async fn main() -> Result<()> {
|
||||||
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.clone())?;
|
let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
|
||||||
let mut players = PlayerRegistry::empty(rooms.clone(), storage.clone(), &mut metrics)?;
|
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 = http::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
|
||||||
let irc = projection_irc::launch(
|
let irc = projection_irc::launch(
|
||||||
irc_config,
|
irc_config,
|
||||||
|
|
Loading…
Reference in New Issue