[code] improve structure (focus: server)

crates/lanspread-server/src/quic.rs

@@ -0,0 +1,152 @@
+use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
+
+use lanspread_db::db::GameDB;
+use lanspread_proto::{Message as _, Request};
+use lanspread_utils::maybe_addr;
+use s2n_quic::{
+    Connection, Server,
+    provider::limits::Limits,
+    stream::{ReceiveStream, SendStream},
+};
+use tokio::io::AsyncWriteExt as _;
+
+use crate::req::{RequestHandler, send_game_file_data};
+
+static KEY_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../key.pem"));
+static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cert.pem"));
+
+#[derive(Clone, Debug)]
+struct ServerCtx {
+    handler: RequestHandler,
+    games_folder: PathBuf,
+}
+
+#[derive(Clone, Debug)]
+struct ConnectionCtx {
+    server_ctx: Arc<ServerCtx>,
+}
+
+#[derive(Clone, Debug)]
+struct StreamCtx {
+    conn_ctx: Arc<ConnectionCtx>,
+}
+
+async fn handle_bidi_stream(
+    mut rx: ReceiveStream,
+    mut tx: SendStream,
+    ctx: Arc<StreamCtx>,
+) -> eyre::Result<()> {
+    let remote_addr = maybe_addr!(rx.connection().remote_addr());
+    tracing::trace!("{remote_addr} stream opened");
+
+    // handle streams
+    while let Ok(Some(data)) = rx.receive().await {
+        tracing::trace!(
+            "{remote_addr} msg: (raw): {}",
+            String::from_utf8_lossy(&data)
+        );
+
+        let request = Request::decode(data);
+        tracing::debug!("{remote_addr} msg: {request:?}");
+
+        // special case for now (send game file data to client)
+        if let Request::GetGameFileData(game_file_desc) = &request {
+            send_game_file_data(
+                game_file_desc,
+                &mut tx,
+                &ctx.conn_ctx.server_ctx.games_folder,
+            )
+            .await;
+            continue;
+        }
+
+        let response = ctx
+            .conn_ctx
+            .server_ctx
+            .handler
+            .handle_request(request, &ctx.conn_ctx.server_ctx.games_folder)
+            .await;
+
+        tracing::trace!("{remote_addr} server response: {response:?}");
+        let raw_response = response.encode();
+        tracing::trace!(
+            "{remote_addr} server response (raw): {}",
+            String::from_utf8_lossy(&raw_response)
+        );
+
+        // write response back to client
+        if let Err(e) = tx.write_all(&raw_response).await {
+            tracing::error!(?e);
+        }
+
+        // close the stream
+        if let Err(e) = tx.close().await {
+            tracing::error!(?e);
+        }
+    }
+
+    Ok(())
+}
+
+async fn handle_connection(
+    mut connection: Connection,
+    ctx: Arc<ConnectionCtx>,
+) -> eyre::Result<()> {
+    let remote_addr = maybe_addr!(connection.remote_addr());
+    tracing::info!("{remote_addr} connected");
+
+    // handle streams
+    while let Ok(Some(stream)) = connection.accept_bidirectional_stream().await {
+        let remote_addr = remote_addr.clone();
+        let (rx, tx) = stream.split();
+
+        let ctx = Arc::new(StreamCtx {
+            conn_ctx: ctx.clone(),
+        });
+
+        // spawn a new task for the stream
+        tokio::spawn(async move {
+            if let Err(e) = handle_bidi_stream(rx, tx, ctx).await {
+                tracing::error!("{remote_addr} stream error: {e}");
+            }
+        });
+    }
+
+    Ok(())
+}
+
+pub(crate) async fn run_server(
+    addr: SocketAddr,
+    db: GameDB,
+    games_folder: PathBuf,
+) -> eyre::Result<()> {
+    let limits = Limits::default()
+        .with_max_idle_timeout(Duration::ZERO)?
+        .with_max_handshake_duration(Duration::from_secs(3))?;
+
+    let mut server = Server::builder()
+        .with_tls((CERT_PEM, KEY_PEM))?
+        .with_io(addr)?
+        .with_limits(limits)?
+        .start()?;
+
+    let server_ctx = Arc::new(ServerCtx {
+        handler: RequestHandler::new(db),
+        games_folder,
+    });
+
+    while let Some(connection) = server.accept().await {
+        let conn_ctx = Arc::new(ConnectionCtx {
+            server_ctx: server_ctx.clone(),
+        });
+
+        // spawn a new task for the connection
+        tokio::spawn(async move {
+            if let Err(e) = handle_connection(connection, conn_ctx).await {
+                tracing::error!("Connection error: {}", e);
+            }
+        });
+    }
+
+    Ok(())
+}