Web Socket & http Server on same port

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",

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"

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<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>,
+    pub sender: Arc<RwLock<SplitSink<WebSocketStream<TokioIo<Upgraded>>, Message>>>,
-    pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
+    pub receiver: Arc<RwLock<SplitStream<WebSocketStream<TokioIo<Upgraded>>>>>,
     pub user_id: Arc<RwLock<Option<Uuid>>>,
     pub community: Arc<RwLock<Option<Arc<Community>>>>,
     identified: Arc<RwLock<bool>>,
@@ -33,8 +38,8 @@ pub struct CommunityConnection {
 }
 impl CommunityConnection {
     pub fn new(
-        sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
+        sender: SplitSink<WebSocketStream<TokioIo<Upgraded>>, Message>,
-        receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
+        receiver: SplitStream<WebSocketStream<TokioIo<Upgraded>>>,
         community: Arc<Community>,
     ) -> Arc<Self> {
         Arc::new(Self {
@@ -50,13 +55,9 @@ impl CommunityConnection {
         })
     }
     pub async fn send_message(&self, message: &CommunicationValue) {
-        let mut session = self.sender.write().await;
+        let mut sender = self.sender.write().await; // Access the SplitSink
-        session
+        let message_text = Message::Text(Utf8Bytes::from(message.to_json().to_string()));
-            .send(Message::Text(Utf8Bytes::from(
+        sender.send(message_text).await.unwrap(); // Send the message via the SplitSink
-                message.to_json().to_string(),
-            )))
-            .await
-            .unwrap();
     }
     pub async fn get_community(&self) -> Option<Arc<Community>> {
         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 mut sender = self.sender.write().await;
-        let _ = session.close(None).await;
+        let _ = sender.close().await;
     }
     pub async fn handle_close(self: Arc<Self>) {
         if self.is_identified().await {

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;
 

src/server/mod.rs

@@ -1 +1,2 @@
+pub mod server;
 pub mod socket;

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<HttpRequest<Incoming>> for HttpService {
+    type Response = HttpResponse<Full<Bytes>>;
+    type Error = io::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(
+        &mut self,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        std::task::Poll::Ready(std::io::Result::Ok(()))
+    }
+
+    fn call(&mut self, req: HttpRequest<Incoming>) -> 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://<host>:<port>/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::<io::Error>())
+                            {
+                                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
+}

src/server/socket.rs

@@ -1,42 +1,30 @@
 use crate::communities::{community_connection::CommunityConnection, community_manager};
 
-use async_tungstenite::{WebSocketStream, accept_hdr_async, tungstenite::protocol::Message};
+use async_tungstenite::accept_hdr_async;
-use futures::{SinkExt, StreamExt};
+use futures::StreamExt;
-use std::{sync::Arc, time::Duration};
+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 async fn start(port: u16) -> bool {
+pub fn handle(
-    let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await;
+    path: String,
-    if let Err(_) = listener {
+    writer: SplitSink<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>, Message>,
-        return false;
+    reader: SplitStream<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>>,
-    }
+) {
-    let listener = listener.unwrap();
     tokio::spawn(async move {
-        while let Ok((stream, _)) = listener.accept().await {
+        if path.starts_with("/ws/community/") {
-            tokio::spawn(async move {
+            let community_id = path.split("/").nth(3).unwrap();
-                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<CommunityConnection> =
-                            Arc::from(CommunityConnection::new(reader, writer, community));
+                    Arc::from(CommunityConnection::new(writer, reader, community));
                 loop {
                     let msg_result = {
                         let mut session_lock = community_conn.receiver.write().await;
@@ -70,6 +58,30 @@ pub async fn start(port: u16) -> bool {
         }
     });
 }
+
+pub async fn start(port: u16) -> bool {
+    let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await;
+    if let Err(_) = listener {
+        return false;
+    }
+    let listener = listener.unwrap();
+    tokio::spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            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
 }