forked from lavina/lavina
1
0
Fork 0
lavina/src/projections/irc/mod.rs

478 lines
16 KiB
Rust
Raw Normal View History

2023-02-07 15:21:00 +00:00
use std::net::SocketAddr;
2023-02-09 19:01:21 +00:00
use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
2023-02-10 18:47:58 +00:00
use tokio::net::tcp::{ReadHalf, WriteHalf};
2023-02-07 15:21:00 +00:00
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::oneshot::channel;
2023-02-13 19:16:00 +00:00
use crate::core::player::{
2023-02-14 18:46:42 +00:00
IncomingPlayerEvent, PlayerCommand, PlayerConnection, PlayerId, PlayerRegistry, Updates,
2023-02-13 19:16:00 +00:00
};
use crate::core::room::{RoomId, RoomInfo, RoomRegistry};
2023-02-07 15:21:00 +00:00
use crate::prelude::*;
2023-02-10 17:09:29 +00:00
use crate::protos::irc::client::{client_message, ClientMessage};
use crate::protos::irc::server::{ServerMessage, ServerMessageBody};
2023-02-13 17:08:37 +00:00
use crate::protos::irc::{Chan, Recipient};
2023-02-07 15:21:00 +00:00
use crate::util::Terminator;
2023-02-13 20:04:08 +00:00
#[cfg(test)]
mod test;
2023-02-10 17:09:29 +00:00
#[derive(Deserialize, Debug, Clone)]
2023-02-07 15:21:00 +00:00
pub struct ServerConfig {
pub listen_on: SocketAddr,
2023-02-10 17:09:29 +00:00
pub server_name: String,
2023-02-07 15:21:00 +00:00
}
2023-02-10 18:47:58 +00:00
#[derive(Debug)]
struct RegisteredUser {
nickname: Vec<u8>,
username: Vec<u8>,
realname: Vec<u8>,
}
2023-02-07 15:21:00 +00:00
async fn handle_socket(
2023-02-10 17:09:29 +00:00
config: ServerConfig,
2023-02-07 15:21:00 +00:00
mut stream: TcpStream,
socket_addr: SocketAddr,
mut players: PlayerRegistry,
rooms: RoomRegistry,
) -> Result<()> {
2023-02-07 15:21:00 +00:00
let (reader, writer) = stream.split();
2023-02-10 18:47:58 +00:00
let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
2023-02-07 15:21:00 +00:00
let mut writer = BufWriter::new(writer);
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::Notice {
first_target: b"*".to_vec(),
rest_targets: vec![],
text: b"Welcome to my server!".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(&mut writer)
.await?;
writer.flush().await?;
2023-02-10 18:47:58 +00:00
let registered_user: Result<RegisteredUser> =
handle_registration(&mut reader, &mut writer).await;
match registered_user {
Ok(user) => {
handle_registered_socket(
config,
socket_addr,
players,
rooms,
&mut reader,
&mut writer,
user,
)
.await?;
}
Err(_) => {}
}
stream.shutdown().await?;
Ok(())
}
2023-02-07 15:21:00 +00:00
async fn handle_registration<'a>(
reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>,
) -> Result<RegisteredUser> {
let mut buffer = vec![];
2023-02-07 15:21:00 +00:00
2023-02-10 18:47:58 +00:00
let mut future_nickname: Option<Vec<u8>> = None;
let mut future_username: Option<(Vec<u8>, Vec<u8>)> = None;
loop {
2023-02-10 18:47:58 +00:00
let res = reader.read_until(b'\n', &mut buffer).await;
match res {
Ok(len) => {
if len == 0 {
log::info!("Terminating socket");
break Err(anyhow::Error::msg("EOF"));
}
}
Err(err) => {
log::warn!("Failed to read from socket: {err}");
break Err(err.into());
}
}
let parsed = client_message(&buffer[..]);
match parsed {
Ok((rest, msg)) => {
log::info!("Incoming IRC message: {msg:?}");
match msg {
ClientMessage::Nick { nickname } => {
if let Some((username, realname)) = future_username {
break Ok(RegisteredUser {
nickname,
username,
realname,
});
} else {
future_nickname = Some(nickname);
}
}
ClientMessage::User { username, realname } => {
if let Some(nickname) = future_nickname {
break Ok(RegisteredUser {
nickname,
username,
realname,
});
} else {
future_username = Some((username, realname));
}
}
_ => {}
}
}
Err(err) => {
log::warn!("Failed to parse IRC message: {err}");
}
}
buffer.clear();
}
}
async fn handle_registered_socket<'a>(
config: ServerConfig,
socket_addr: SocketAddr,
mut players: PlayerRegistry,
rooms: RoomRegistry,
reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>,
2023-02-10 18:47:58 +00:00
user: RegisteredUser,
) -> Result<()> {
2023-02-10 18:47:58 +00:00
let mut buffer = vec![];
log::info!("Handling registered user: {user:?}");
2023-02-13 20:58:44 +00:00
let player_id = PlayerId(user.nickname.clone());
let mut connection = players.connect_to_player(player_id.clone()).await;
2023-02-12 23:31:16 +00:00
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N001Welcome {
client: user.nickname.clone(),
text: b"Welcome to Kek Server".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N002YourHost {
client: user.nickname.clone(),
text: b"Welcome to Kek Server".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N003Created {
client: user.nickname.clone(),
text: b"Welcome to Kek Server".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N004MyInfo {
client: user.nickname.clone(),
hostname: config.server_name.as_bytes().to_vec(),
softname: b"kek-0.1.alpha.3".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N005ISupport {
client: user.nickname.clone(),
params: b"CHANTYPES=#".to_vec(),
},
2023-02-10 18:47:58 +00:00
}
.write_async(writer)
.await?;
2023-02-14 18:28:49 +00:00
let (promise, deferred) = oneshot();
connection.send(PlayerCommand::GetRooms(promise)).await;
let rooms_list = deferred.await?;
for room in &rooms_list {
2023-02-14 18:46:42 +00:00
produce_on_join_cmd_messages(
&config,
&user,
&Chan::Global(room.id.0.clone()),
room,
writer,
)
.await?;
2023-02-14 18:28:49 +00:00
}
writer.flush().await?;
2023-02-10 18:47:58 +00:00
2023-02-07 15:21:00 +00:00
loop {
select! {
biased;
len = reader.read_until(b'\n', &mut buffer) => {
let len = len?;
let len = if len == 0 {
log::info!("EOF, Terminating socket");
break;
} else {
len
};
2023-02-14 18:46:42 +00:00
handle_incoming_message(&buffer[0..len], &config, &user, &rooms, &mut connection, writer).await?;
2023-02-07 15:21:00 +00:00
buffer.clear();
},
2023-02-13 19:16:00 +00:00
update = connection.receiver.recv() => {
2023-02-13 17:08:37 +00:00
match update.unwrap() {
Updates::RoomJoined { player_id: author_id, connection_id, room_id } => {
if player_id == author_id {
if let Some(room) = rooms.get_room(&room_id) {
let room_info = room.get_room_info().await;
let chan = Chan::Global(room_id.0);
produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?;
writer.flush().await?;
} else {
log::warn!("Received join to a non-existant room");
}
} else {
ServerMessage {
tags: vec![],
sender: Some(author_id.0.clone()),
body: ServerMessageBody::Join(Chan::Global(room_id.0)),
}.write_async(writer).await?;
writer.flush().await?
}
},
2023-02-13 18:32:52 +00:00
Updates::NewMessage { author_id, connection_id, room_id, body } => {
2023-02-13 20:58:44 +00:00
if player_id != author_id || connection.connection_id != connection_id {
2023-02-13 18:32:52 +00:00
ServerMessage {
tags: vec![],
sender: Some(author_id.0.clone()),
body: ServerMessageBody::PrivateMessage { target: Recipient::Chan(Chan::Global(room_id.0)), body: body.as_bytes().to_vec() }
}.write_async(writer).await?;
2023-02-13 18:32:52 +00:00
writer.flush().await?
}
2023-02-13 17:08:37 +00:00
},
2023-02-14 18:46:42 +00:00
Updates::IncomingUpdate(IncomingPlayerEvent::IncomingRoomTopicChanged { room_id, new_topic }) => {
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
chat: Chan::Global(room_id.0.clone()),
topic: new_topic,
},
}.write_async(writer).await?;
writer.flush().await?
},
2023-02-13 17:08:37 +00:00
}
}
2023-02-07 15:21:00 +00:00
}
}
Ok(())
2023-02-07 15:21:00 +00:00
}
async fn handle_incoming_message(
buffer: &[u8],
config: &ServerConfig,
user: &RegisteredUser,
2023-02-14 18:46:42 +00:00
rooms: &RoomRegistry,
2023-02-13 19:16:00 +00:00
user_handle: &mut PlayerConnection,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
let parsed = client_message(buffer);
match parsed {
Ok((rest, msg)) => match msg {
ClientMessage::Ping { token } => {
ServerMessage {
tags: vec![],
sender: None,
body: ServerMessageBody::Pong {
from: config.server_name.as_bytes().to_vec(),
token,
},
}
.write_async(writer)
.await?;
writer.flush().await?;
}
ClientMessage::Join(ref chan) => {
handle_join(&config, &user, user_handle, chan, writer).await?;
}
ClientMessage::PrivateMessage { recipient, body } => match recipient {
Recipient::Chan(Chan::Global(room)) => match String::from_utf8(body) {
Ok(body) => {
user_handle
2023-02-13 19:16:00 +00:00
.send_message(RoomId(room.clone()), body.clone())
.await
}
Err(err) => log::warn!("failed to parse incoming message: {err}"),
},
_ => log::warn!("Unsupported target type"),
},
2023-02-14 18:46:42 +00:00
ClientMessage::Topic { chan, topic } => {
match chan {
Chan::Global(ref room) => {
let room = rooms.get_room(&RoomId(room.clone()));
if let Some(mut room) = room {
room.set_topic(topic).await;
}
}
Chan::Local(_) => {}
};
}
_ => {}
},
Err(err) => {
log::warn!("Failed to parse IRC message: {err}");
}
}
Ok(())
}
async fn handle_join(
config: &ServerConfig,
user: &RegisteredUser,
2023-02-13 19:16:00 +00:00
user_handle: &mut PlayerConnection,
chan: &Chan,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
match chan {
Chan::Global(ref room) => {
2023-02-14 00:44:03 +00:00
let room_info = user_handle.join_room(RoomId(room.clone())).await?;
}
Chan::Local(_) => {}
};
Ok(())
}
async fn produce_on_join_cmd_messages(
config: &ServerConfig,
user: &RegisteredUser,
chan: &Chan,
room_info: &RoomInfo,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
ServerMessage {
tags: vec![],
sender: Some(user.nickname.clone()),
body: ServerMessageBody::Join(chan.clone()),
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
chat: chan.clone(),
topic: room_info.topic.clone(),
},
}
.write_async(writer)
.await?;
let mut members = if let Some(head) = room_info.members.first() {
head.0.clone()
} else {
user.nickname.clone()
};
for i in &room_info.members[1..] {
members.push(b' ');
members.extend(&i.0);
}
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N353NamesReply {
client: user.nickname.clone(),
chan: chan.clone(),
members,
},
}
.write_async(writer)
.await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.as_bytes().to_vec()),
body: ServerMessageBody::N366NamesReplyEnd {
client: user.nickname.clone(),
chan: chan.clone(),
},
}
.write_async(writer)
.await?;
Ok(())
}
2023-02-09 19:01:21 +00:00
pub async fn launch(
2023-02-10 17:09:29 +00:00
config: ServerConfig,
2023-02-09 19:01:21 +00:00
players: PlayerRegistry,
rooms: RoomRegistry,
2023-02-09 19:01:21 +00:00
metrics: MetricsRegistry,
) -> Result<Terminator> {
2023-02-07 15:21:00 +00:00
log::info!("Starting IRC projection");
let (signal, mut rx) = channel();
2023-02-09 19:01:21 +00:00
let current_connections =
IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
let total_connections = IntCounter::new(
"irc_total_connections",
"Total number of opened connections",
)?;
metrics.register(Box::new(current_connections.clone()))?;
metrics.register(Box::new(total_connections.clone()))?;
2023-02-07 15:21:00 +00:00
let listener = TcpListener::bind(config.listen_on).await?;
log::debug!("Listener started");
let handle = tokio::task::spawn(async move {
loop {
select! {
biased;
_ = &mut rx => break,
new_conn = listener.accept() => {
match new_conn {
Ok((stream, socket_addr)) => {
2023-02-10 17:09:29 +00:00
let config = config.clone();
2023-02-09 19:01:21 +00:00
total_connections.inc();
current_connections.inc();
2023-02-07 15:21:00 +00:00
log::debug!("Incoming connection from {socket_addr}");
let players = players.clone();
let rooms = rooms.clone();
let current_connections_clone = current_connections.clone();
let handle = tokio::task::spawn(async move {
match handle_socket(config, stream, socket_addr, players, rooms).await {
Ok(_) => log::info!("Connection terminated"),
Err(err) => log::warn!("Connection failed: {err}"),
}
current_connections_clone.dec();
});
2023-02-07 15:21:00 +00:00
},
Err(err) => log::warn!("Failed to accept new connection: {err}"),
}
},
}
}
log::info!("Stopping IRC projection");
Ok(())
});
let terminator = Terminator::from_raw(signal, handle);
Ok(terminator)
}