From 78a0ebd18e44c5051b1f78531ea27b24ac9fce4b Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Mon, 10 Nov 2025 22:12:38 +0000 Subject: [PATCH] Web Socket & http Server on same port --- Cargo.lock | 3 + Cargo.toml | 3 + src/communities/community_connection.rs | 29 ++--- src/main.rs | 3 +- src/server/mod.rs | 1 + src/server/server.rs | 160 ++++++++++++++++++++++++ src/server/socket.rs | 120 ++++++++++-------- 7 files changed, 250 insertions(+), 69 deletions(-) create mode 100644 src/server/server.rs diff --git a/Cargo.lock b/Cargo.lock index bbdc5cb..a76b5a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -21,7 +21,9 @@ dependencies = [ "hex", "hkdf", "hmac", + "http-body-util", "hyper", + "hyper-util", "json", "native-tls", "once_cell", @@ -39,6 +41,7 @@ dependencies = [ "tokio", "tokio-tungstenite", "tokio-util", + "tower", "tungstenite", "uuid", "walkdir", diff --git a/Cargo.toml b/Cargo.toml index 3bf462d..d7d1300 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,8 @@ hex = "*" hkdf = "*" hmac = "*" hyper = { version = "*", features = ["full"] } +hyper-util = "*" +http-body-util = "*" json = "*" native-tls = { version = "*", default-features = false } once_cell = "1.21.3" @@ -37,6 +39,7 @@ sysinfo = "0.30" tokio = { version = "*", features = ["full"] } tokio-util = { version = "*", features = ["full"] } tokio-tungstenite = { version = "*", features = ["native-tls"] } +tower = "*" tungstenite = "*" uuid = { version = "*", features = ["v4"] } walkdir = "2.5.0" diff --git a/src/communities/community_connection.rs b/src/communities/community_connection.rs index 85ef69d..20f816c 100644 --- a/src/communities/community_connection.rs +++ b/src/communities/community_connection.rs @@ -6,23 +6,28 @@ use crate::data::communication::{CommunicationType, CommunicationValue, DataType use aes_gcm::{Aes256Gcm, KeyInit, Nonce, aead::Aead}; use async_tungstenite::WebSocketReceiver; use async_tungstenite::WebSocketSender; +use async_tungstenite::tungstenite::Message; use async_tungstenite::tungstenite::Utf8Bytes; -use async_tungstenite::{WebSocketStream, tungstenite::Message}; use base64::{Engine as _, engine::general_purpose::STANDARD}; use futures::SinkExt; +use futures::stream::SplitSink; +use futures::stream::SplitStream; use hkdf::Hkdf; +use hyper::upgrade::Upgraded; +use hyper_util::rt::TokioIo; use json::JsonValue; use rand::{Rng, distributions::Alphanumeric}; use sha2::Sha256; use std::sync::Arc; use tokio::sync::Mutex; use tokio::sync::RwLock; +use tokio_tungstenite::WebSocketStream; use tokio_util::compat::Compat; use uuid::Uuid; use x448::PublicKey; pub struct CommunityConnection { - pub sender: Arc>>>, - pub receiver: Arc>>>, + pub sender: Arc>, Message>>>, + pub receiver: Arc>>>>, pub user_id: Arc>>, pub community: Arc>>>, identified: Arc>, @@ -33,8 +38,8 @@ pub struct CommunityConnection { } impl CommunityConnection { pub fn new( - sender: WebSocketSender>, - receiver: WebSocketReceiver>, + sender: SplitSink>, Message>, + receiver: SplitStream>>, community: Arc, ) -> Arc { Arc::new(Self { @@ -50,13 +55,9 @@ impl CommunityConnection { }) } pub async fn send_message(&self, message: &CommunicationValue) { - let mut session = self.sender.write().await; - session - .send(Message::Text(Utf8Bytes::from( - message.to_json().to_string(), - ))) - .await - .unwrap(); + let mut sender = self.sender.write().await; // Access the SplitSink + let message_text = Message::Text(Utf8Bytes::from(message.to_json().to_string())); + sender.send(message_text).await.unwrap(); // Send the message via the SplitSink } pub async fn get_community(&self) -> Option> { self.community.read().await.clone() @@ -398,8 +399,8 @@ impl CommunityConnection { self.send_message(&error).await; } pub async fn close(&self) { - let mut session = self.sender.write().await; - let _ = session.close(None).await; + let mut sender = self.sender.write().await; + let _ = sender.close().await; } pub async fn handle_close(self: Arc) { if self.is_identified().await { diff --git a/src/main.rs b/src/main.rs index 686d2ba..abcac06 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,7 @@ mod server; mod users; mod util; +use crate::communities::community::Community; use crate::communities::community_manager; use crate::communities::interactables::registry; use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes}; @@ -25,7 +26,7 @@ use crate::gui::{log_panel, ratatui_interface}; use crate::langu::language_creator; use crate::langu::language_manager::format; use crate::omikron::omikron_connection::OmikronConnection; -use crate::server::socket::start; +use crate::server::server::start; use crate::users::user_manager; use crate::util::config_util::CONFIG; diff --git a/src/server/mod.rs b/src/server/mod.rs index d22cc84..538dc53 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1 +1,2 @@ +pub mod server; pub mod socket; diff --git a/src/server/server.rs b/src/server/server.rs new file mode 100644 index 0000000..f017a9c --- /dev/null +++ b/src/server/server.rs @@ -0,0 +1,160 @@ +use color_eyre::eyre::Ok; +use futures::{StreamExt, TryFutureExt}; +use http_body_util::Full; +use hyper::body::Bytes; +use hyper::{ + Request as HttpRequest, Response as HttpResponse, StatusCode, body::Incoming, + server::conn::http1, upgrade, +}; +use std::error::Error; +use std::io::{self}; +use std::{future::Future, pin::Pin, time::Duration}; +use tokio::net::TcpListener; +use tower::Service; +use warp::filters::log::log; + +use crate::gui::log_panel::log_message; +use crate::langu::language_manager::format; +use crate::server::socket::handle; +use hyper_util::rt::tokio::TokioIo; +use hyper_util::service::TowerToHyperService; +use tokio_tungstenite::WebSocketStream; + +#[derive(Clone)] +struct HttpService; + +impl Service> for HttpService { + type Response = HttpResponse>; + type Error = io::Error; + type Future = Pin> + Send>>; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(std::io::Result::Ok(())) + } + + fn call(&mut self, req: HttpRequest) -> Self::Future { + let path = req.uri().path().to_string(); + let headers = req.headers().clone(); + let upgrades = upgrade::on(req); + + let fut = async move { + if path.starts_with("/ws") + && headers + .get("connection") + .map(|v| v.to_str().unwrap_or("").contains("Upgrade")) + .unwrap_or(false) + && headers.get("upgrade").map(|v| v.to_str().unwrap_or("")) == Some("websocket") + { + log_message("Attempting WebSocket upgrade on /ws"); + + let response = HttpResponse::builder() + .status(StatusCode::SWITCHING_PROTOCOLS) + .header("Upgrade", "websocket") + .header("Connection", "Upgrade") + .body(Full::new(Bytes::from(""))) + .unwrap(); + tokio::spawn(async move { + match upgrades.await { + std::result::Result::Ok(upgraded_stream) => { + let raw_stream = TokioIo::new(upgraded_stream); + + let handshake_result = WebSocketStream::from_raw_socket( + raw_stream, + tungstenite::protocol::Role::Server, + None, + ) + .await; + let (writer, reader) = handshake_result.split(); + handle(path, writer, reader); + } + Err(e) => { + log_message(format!( + "WebSocket upgrade failed after response: {:?}", + e + )); + } + } + }); + Ok(response) + } else { + let (status, body_text) = match path.as_str() { + "/" => ( + StatusCode::OK, + "Barebones Server: Try connecting to WebSocket at ws://:/ws or check /status.", + ), + "/status" => (StatusCode::OK, "HTTP Server Status: Barebones Online"), + _ => (StatusCode::NOT_FOUND, "404 Not Found"), + }; + let body = Full::new(Bytes::from(body_text.to_string())); + let response = HttpResponse::builder() + .status(status) + .header(hyper::header::CONTENT_TYPE, "text/plain") + .body(body) + .unwrap(); + Ok(response) + } + }; + + Box::pin(fut.map_err(|err| { + io::Error::new( + io::ErrorKind::Other, + format!("Error in request handling: {}", err), + ) + })) + } +} + +pub async fn start(port: u16) -> bool { + // Bind to the port + let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await; + if let Err(e) = listener { + log_message(format!("Failed to bind to port {}: {:?}", port, e)); + return false; + } + let listener = listener.unwrap(); + log_message(format!( + "Barebones Server listening for HTTP and WS on 0.0.0.0:{}", + port + )); + + tokio::spawn(async move { + loop { + match listener.accept().await { + std::result::Result::Ok((stream, _addr)) => { + let service = HttpService; + tokio::spawn(async move { + let io = TokioIo::new(stream); + + if let Err(err) = http1::Builder::new() + .preserve_header_case(true) + .title_case_headers(true) + .serve_connection(io, TowerToHyperService::new(service)) + .with_upgrades() + .await + { + if let Some(io_err) = + err.source().and_then(|e| e.downcast_ref::()) + { + if io_err.kind() != io::ErrorKind::ConnectionReset + && io_err.kind() != io::ErrorKind::BrokenPipe + { + log_message(format!("Error serving connection: {:?}", err)); + } + } else { + log_message(format!("Error serving connection: {:?}", err)); + } + } + }); + } + Err(e) => { + log_message(format!("Error accepting connection: {:?}", e)); + tokio::time::sleep(Duration::from_millis(500)).await; + } + } + } + }); + true +} diff --git a/src/server/socket.rs b/src/server/socket.rs index 58a1edf..627e165 100644 --- a/src/server/socket.rs +++ b/src/server/socket.rs @@ -1,16 +1,64 @@ use crate::communities::{community_connection::CommunityConnection, community_manager}; -use async_tungstenite::{WebSocketStream, accept_hdr_async, tungstenite::protocol::Message}; -use futures::{SinkExt, StreamExt}; -use std::{sync::Arc, time::Duration}; +use async_tungstenite::accept_hdr_async; +use futures::StreamExt; +use futures::stream::SplitSink; +use futures::stream::SplitStream; +use hyper::upgrade::Upgraded; +use hyper_util::rt::TokioIo; +use std::sync::Arc; use tokio::net::TcpListener; use tokio_util::compat::{Compat, TokioAsyncReadCompatExt}; -use tungstenite::connect; use tungstenite::{ - Utf8Bytes, + Message, Utf8Bytes, handshake::server::{Request, Response}, }; +pub fn handle( + path: String, + writer: SplitSink>, Message>, + reader: SplitStream>>, +) { + tokio::spawn(async move { + if path.starts_with("/ws/community/") { + let community_id = path.split("/").nth(3).unwrap(); + if let Some(community) = community_manager::get_community(community_id).await { + let community_conn: Arc = + Arc::from(CommunityConnection::new(writer, reader, community)); + loop { + let msg_result = { + let mut session_lock = community_conn.receiver.write().await; + session_lock.next().await + }; + + match msg_result { + Some(Ok(msg)) => { + if msg.is_text() { + let text = msg.into_text().unwrap(); + community_conn + .clone() + .handle_message(text.to_string()) + .await; + } else if msg.is_close() { + community_conn.handle_close().await; + return; + } + } + Some(Err(_)) => { + community_conn.handle_close().await; + return; + } + None => { + community_conn.handle_close().await; + return; + } + } + } + } + } + }); +} + pub async fn start(port: u16) -> bool { let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await; if let Err(_) = listener { @@ -19,56 +67,20 @@ pub async fn start(port: u16) -> bool { let listener = listener.unwrap(); tokio::spawn(async move { while let Ok((stream, _)) = listener.accept().await { - tokio::spawn(async move { - let mut path: String = "/".to_string(); - let callback = |req: &Request, response: Response| { - path = format!("{}", &req.uri().path()); - Ok(response) - }; - let ws_stream = match accept_hdr_async(stream.compat(), callback).await { - Ok(ws) => ws, - Err(_) => { - return; - } - }; - let (reader, writer) = ws_stream.split(); - if path.starts_with("/community/") { - let community_id = path.split("/").nth(2).unwrap(); - if let Some(community) = community_manager::get_community(community_id).await { - let community_conn: Arc = - Arc::from(CommunityConnection::new(reader, writer, community)); - loop { - let msg_result = { - let mut session_lock = community_conn.receiver.write().await; - session_lock.next().await - }; - - match msg_result { - Some(Ok(msg)) => { - if msg.is_text() { - let text = msg.into_text().unwrap(); - community_conn - .clone() - .handle_message(text.to_string()) - .await; - } else if msg.is_close() { - community_conn.handle_close().await; - return; - } - } - Some(Err(_)) => { - community_conn.handle_close().await; - return; - } - None => { - community_conn.handle_close().await; - return; - } - } - } - } + let mut path: String = "/".to_string(); + let callback = |req: &Request, response: Response| { + path = format!("{}", &req.uri().path()); + Ok(response) + }; + let ws_stream = match accept_hdr_async(stream.compat(), callback).await { + Ok(ws) => ws, + Err(_) => { + return; } - }); + }; + + let (reader, writer) = ws_stream.split(); + //handle(path, reader, writer); } }); true