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

Compare commits

base: nikita:041c1b5fe2a12fc45c3123552cd0632289a24a00
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:36b0d50d5103f6bd86ef52680ce7f001ae3b1d7d
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

No commits in common. "041c1b5fe2a12fc45c3123552cd0632289a24a00" and "36b0d50d5103f6bd86ef52680ce7f001ae3b1d7d" have entirely different histories.
041c1b5fe2 ... 36b0d50d51

16 changed files with 108 additions and 370 deletions
Show Stats Expand all files Collapse all files

21
.pre-commit-config.yaml
View File

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

68
Cargo.lock generated
View File

@ -45,21 +45,6 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "anstream"
version = "0.6.13"
@ -231,20 +216,6 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "clap"
version = "4.5.3"
@ -303,12 +274,6 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
[[package]]
name = "core-foundation-sys"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f"
[[package]]
name = "cpufeatures"
version = "0.2.12"
@ -764,29 +729,6 @@ dependencies = [
"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]]
name = "idna"
version = "0.5.0"
@ -876,7 +818,6 @@ name = "lavina-core"
version = "0.0.2-dev"
dependencies = [
"anyhow",
"chrono",
"prometheus",
"serde",
"sqlx",
@ -2442,15 +2383,6 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]]
name = "windows-sys"
version = "0.48.0"

1
crates/lavina-core/Cargo.toml
View File

@ -10,4 +10,3 @@ serde.workspace = true
tokio.workspace = true
tracing.workspace = true
prometheus.workspace = true
chrono = "0.4.37"

1
crates/lavina-core/migrations/2_created_at_for_messages.sql
View File

@ -1 +0,0 @@
alter table messages add column created_at text;

56
crates/lavina-core/src/player.rs
View File

@ -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,
//! so that they don't overload the room actor.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::{
collections::{HashMap, HashSet},
sync::{Arc, RwLock},
};
use prometheus::{IntGauge, Registry as MetricsRegistry};
use serde::Serialize;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::RwLock;
use crate::prelude::*;
use crate::repo::Storage;
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
use crate::table::{AnonTable, Key as AnonKey};
@ -208,28 +208,23 @@ pub enum Updates {
#[derive(Clone)]
pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry {
pub fn empty(
room_registry: RoomRegistry,
storage: Storage,
metrics: &mut MetricsRegistry,
) -> Result<PlayerRegistry> {
pub fn empty(room_registry: RoomRegistry, metrics: &mut MetricsRegistry) -> Result<PlayerRegistry> {
let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?;
metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner {
room_registry,
storage,
players: HashMap::new(),
metric_active_players,
};
Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
}
pub async fn get_or_launch_player(&mut self, id: PlayerId) -> PlayerHandle {
let mut inner = self.0.write().await;
pub async fn get_or_create_player(&mut self, id: PlayerId) -> PlayerHandle {
let mut inner = self.0.write().unwrap();
if let Some((handle, _)) = inner.players.get(&id) {
handle.clone()
} 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.metric_active_players.inc();
handle
@ -237,12 +232,12 @@ impl PlayerRegistry {
}
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
}
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() {
k.send(ActorCommand::Stop).await;
drop(k);
@ -257,8 +252,6 @@ impl PlayerRegistry {
/// The player registry state representation.
struct PlayerRegistryInner {
room_registry: RoomRegistry,
storage: Storage,
/// Active player actors.
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
metric_active_players: IntGauge,
}
@ -266,31 +259,26 @@ struct PlayerRegistryInner {
/// Player actor inner state representation.
struct Player {
player_id: PlayerId,
storage_id: u32,
connections: AnonTable<Sender<Updates>>,
my_rooms: HashMap<RoomId, RoomHandle>,
banned_from: HashSet<RoomId>,
rx: Receiver<ActorCommand>,
handle: PlayerHandle,
rooms: RoomRegistry,
storage: Storage,
}
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 handle = PlayerHandle { tx };
let handle_clone = handle.clone();
let storage_id = storage.retrieve_user_id_by_name(player_id.as_inner()).await.unwrap().unwrap();
let player = Player {
player_id,
storage_id,
connections: AnonTable::new(),
my_rooms: HashMap::new(),
banned_from: HashSet::new(),
banned_from: HashSet::from([RoomId::from("Empty").unwrap()]),
rx,
handle,
rooms,
storage,
};
let fiber = tokio::task::spawn(player.main_loop());
(handle_clone, fiber)
@ -384,7 +372,6 @@ impl Player {
todo!();
}
};
room.add_member(self.player_id.clone(), self.storage_id).await;
room.subscribe(self.player_id.clone(), self.handle.clone()).await;
self.my_rooms.insert(room_id.clone(), room.clone());
let room_info = room.get_room_info().await;
@ -400,7 +387,6 @@ impl Player {
let room = self.my_rooms.remove(&room_id);
if let Some(room) = room {
room.unsubscribe(&self.player_id).await;
room.remove_member(&self.player_id).await;
}
let update = Updates::RoomLeft {
room_id,
@ -410,11 +396,12 @@ impl Player {
}
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");
return;
};
room.send_message(self.player_id.clone(), body.clone()).await;
}
let update = Updates::NewMessage {
room_id,
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) {
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");
return;
};
room.set_topic(self.player_id.clone(), new_topic.clone()).await;
}
let update = Updates::RoomTopicChanged { room_id, new_topic };
self.broadcast_update(update, connection_id).await;
}

8
crates/lavina-core/src/repo/mod.rs
View File

@ -11,9 +11,6 @@ use tokio::sync::Mutex;
use crate::prelude::*;
mod user;
mod room;
#[derive(Deserialize, Debug, Clone)]
pub struct StorageConfig {
pub db_path: String,
@ -90,15 +87,14 @@ impl Storage {
return Err(anyhow!("No such user"));
};
sqlx::query(
"insert into messages(room_id, id, content, author_id, created_at)
values (?, ?, ?, ?, ?);
"insert into messages(room_id, id, content, author_id)
values (?, ?, ?, ?);
update rooms set message_count = message_count + 1 where id = ?;",
)
.bind(room_id)
.bind(id)
.bind(content)
.bind(author_id)
.bind(chrono::Utc::now().to_string())
.bind(room_id)
.execute(&mut *executor)
.await?;

19
crates/lavina-core/src/repo/room.rs
View File

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

15
crates/lavina-core/src/repo/user.rs
View File

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

39
crates/lavina-core/src/room.rs
View File

@ -1,5 +1,4 @@
//! Domain of rooms — chats with multiple participants.
use std::collections::HashSet;
use std::{collections::HashMap, hash::Hash, sync::Arc};
use prometheus::{IntGauge, Registry as MetricRegistry};
@ -60,7 +59,6 @@ impl RoomRegistry {
storage_id: stored_room.id,
room_id: room_id.clone(),
subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions
members: HashSet::new(), // TODO load members from storage
topic: stored_room.topic.into(),
message_count: stored_room.message_count,
storage: inner.storage.clone(),
@ -78,7 +76,6 @@ impl RoomRegistry {
storage_id: id,
room_id: room_id.clone(),
subscriptions: HashMap::new(),
members: HashSet::new(),
topic: topic.into(),
message_count: 0,
storage: inner.storage.clone(),
@ -121,32 +118,12 @@ pub struct RoomHandle(Arc<AsyncRwLock<Room>>);
impl RoomHandle {
pub async fn subscribe(&self, player_id: PlayerId, player_handle: PlayerHandle) {
let mut lock = self.0.write().await;
tracing::info!("Adding a subscriber to a room");
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;
lock.add_subscriber(player_id, player_handle).await;
}
pub async fn unsubscribe(&self, player_id: &PlayerId) {
let mut lock = self.0.write().await;
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 {
room_id: lock.room_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;
lock.topic = new_topic.clone();
let update = Updates::RoomTopicChanged {
@ -189,14 +166,22 @@ struct Room {
room_id: RoomId,
/// Player actors on the local node which are subscribed to this room's updates.
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.
message_count: u32,
topic: Str,
storage: Storage,
}
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<()> {
tracing::info!("Adding a message to room");
self.storage.insert_message(self.storage_id, self.message_count, &body, &*author_id.as_inner()).await?;

102
crates/projection-irc/tests/lib.rs
View File

@ -111,36 +111,7 @@ impl TestServer {
})
.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,
})
}
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 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,
@ -181,76 +152,6 @@ async fn scenario_basic() -> Result<()> {
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]
async fn scenario_force_join_msg() -> Result<()> {
let mut server = TestServer::start().await?;
@ -506,6 +407,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
#[tokio::test]
async fn terminate_socket_scenario() -> Result<()> {
let mut server = TestServer::start().await?;
let address: SocketAddr = ("127.0.0.1:0".parse().unwrap());
// test scenario

4
crates/projection-xmpp/src/iq.rs
View File

@ -24,9 +24,9 @@ impl<'a> XmppConnection<'a> {
to: None,
r#type: IqType::Result,
body: BindResponse(Jid {
name: Some(self.user.xmpp_name.clone()),
name: Some(Name("darova".into())),
server: Server("localhost".into()),
resource: Some(self.user.xmpp_resource.clone()),
resource: Some(Resource("kek".into())),
}),
};
req.serialize(output);

20
crates/projection-xmpp/src/lib.rs
View File

@ -52,17 +52,9 @@ struct LoadedConfig {
}
struct Authenticated {
/// Identifier of the authenticated player.
///
/// Used when communicating with lavina-core on behalf of the player.
player_id: PlayerId,
/// The user's XMPP name.
///
/// Used in `to` and `from` fields of XMPP messages.
xmpp_name: Name,
/// The resource given to this user by the server.
xmpp_resource: Resource,
/// The resource used by this user when joining MUCs.
xmpp_muc_name: Resource,
}
@ -315,13 +307,11 @@ async fn socket_auth(
return Err(fail("passwords do not match"));
}
let name: Str = name.as_str().into();
Ok(Authenticated {
player_id: PlayerId::from(name.clone())?,
xmpp_name: Name(name.clone()),
xmpp_resource: Resource(name.clone()),
xmpp_muc_name: Resource(name.clone()),
player_id: PlayerId::from(name.as_str())?,
xmpp_name: Name(name.to_string().into()),
xmpp_resource: Resource(name.to_string().into()),
xmpp_muc_name: Resource(name.to_string().into()),
})
}
Err(e) => return Err(e),
@ -372,7 +362,7 @@ async fn socket_final(
res = &mut next_xml_event => 's: {
let (ns, event) = res?;
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;
}
}

112
crates/projection-xmpp/tests/lib.rs
View File

@ -20,7 +20,7 @@ use tokio_rustls::TlsConnector;
use lavina_core::player::PlayerRegistry;
use lavina_core::repo::{Storage, StorageConfig};
use lavina_core::room::RoomRegistry;
use projection_xmpp::{launch, RunningServer, ServerConfig};
use projection_xmpp::{launch, ServerConfig};
use proto_xmpp::xml::{Continuation, FromXml, Parser};
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]
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
server.storage.create_user("tester").await?;
server.storage.set_password("tester", "password").await?;
storage.create_user("tester").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);
tracing::info!("TCP connection established");
@ -189,7 +169,7 @@ async fn scenario_basic() -> Result<()> {
.with_no_client_auth(),
));
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");
let mut s = TestScopeTls::new(&mut stream, buffer);
@ -203,20 +183,33 @@ async fn scenario_basic() -> Result<()> {
// wrap up
server.server.terminate().await?;
server.terminate().await?;
Ok(())
}
#[tokio::test]
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
server.storage.create_user("tester").await?;
server.storage.set_password("tester", "password").await?;
storage.create_user("tester").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);
tracing::info!("TCP connection established");
@ -240,7 +233,7 @@ async fn scenario_basic_without_headers() -> Result<()> {
.with_no_client_auth(),
));
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");
let mut s = TestScopeTls::new(&mut stream, buffer);
@ -253,20 +246,33 @@ async fn scenario_basic_without_headers() -> Result<()> {
// wrap up
server.server.terminate().await?;
server.terminate().await?;
Ok(())
}
#[tokio::test]
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
server.storage.create_user("tester").await?;
server.storage.set_password("tester", "password").await?;
storage.create_user("tester").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);
tracing::info!("TCP connection established");
@ -292,10 +298,10 @@ async fn terminate_socket() -> Result<()> {
));
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");
server.server.terminate().await?;
server.terminate().await?;
assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);

3
crates/proto-xmpp/src/bind.rs
View File

@ -11,15 +11,12 @@ pub const XMLNS: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
// 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)]
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)]
pub struct Server(pub Str);
/// Resource of an XMPP entity. Placed after the `/` in a JID.
#[derive(PartialEq, Eq, Debug, Clone)]
pub struct Resource(pub Str);

5
docs/cheatsheet.md
View File

@ -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.
Generate self-signed TLS certificate. Mind the common name (CN) field, it should match the domain name of the server.
Example for localhost:
Generate self-signed TLS certificate:
openssl req -x509 -newkey rsa:4096 -sha256 -days 365 -noenc \
-keyout certs/xmpp.key -out certs/xmpp.pem \
-subj "/CN=localhost"
-subj "/CN=example.com"
Print content of a TLS certificate:

2
src/main.rs
View File

@ -52,7 +52,7 @@ async fn main() -> Result<()> {
let mut metrics = MetricsRegistry::new();
let storage = Storage::open(storage_config).await?;
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 irc = projection_irc::launch(
irc_config,