feat: Rust Server Implementation

Author: bsmithcompsci

Cargo.toml

@@ -0,0 +1,2 @@
+[workspace]
+members = ["demo/server/rust-example"]

demo/server/README.md

@@ -0,0 +1,5 @@
+# Server Demos
+
+These servers are built to reflect the various languages to setup an edge service.
+
+An Edge service's idea is to simply and solely connect a player's connection to a game server via kafka.

demo/server/rust-example/.gitignore

@@ -0,0 +1,18 @@
+# BEGIN: Rust
+# Generated by Cargo
+# https://git.io/JvCJW
+
+# Compiled files
+/target/
+
+# Dependency files
+Cargo.lock
+
+# Generated files
+Cargo.toml.orig
+Cargo.toml.bkp
+
+# Output files
+*.rs.bk
+
+# END: Rust

demo/server/rust-example/Cargo.toml

@@ -2,5 +2,19 @@
 name = "rust-example"
 version = "0.1.0"
 edition = "2021"
+publish = false
 
 [dependencies]
+axum = { version = "0.7", features = ["ws", "tokio"] }
+# axum-server = { version = "0.7", features = ["tls-rustls"] }
+axum-extra = { version = "0.9.3", features = ["typed-header"] }
+futures = "0.3"
+futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
+headers = "0.4"
+tokio = { version = "1.4", features = ["full"] }
+tokio-tungstenite = "0.23"
+tower = { version = "0.5", features = ["util"] }
+tower-http = { version = "0.5", features = ["fs", "trace"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+rdkafka = { version = "0.36.2", features = ["cmake-build"] }

demo/server/rust-example/src/api/client.rs

@@ -0,0 +1,20 @@
+pub struct Client
+{
+    id: i64,
+}
+
+impl Client
+{
+    pub fn new(id: i64) -> Client
+    {
+        Client
+        {
+            id,
+        }
+    }
+
+    pub fn get_id(&self) -> i64
+    {
+        self.id
+    }
+}

demo/server/rust-example/src/api/loadbalancer.rs

@@ -0,0 +1,3 @@
+pub trait LoadBalancer {
+    fn next(&mut self) -> &mut super::server::Server;
+}

demo/server/rust-example/src/api/loadbalancers/leastconnections.rs

@@ -0,0 +1,19 @@
+use crate::api::{loadbalancer::LoadBalancer, server::Server};
+
+pub struct LeastConnections {
+    servers: Vec<Server>
+}
+
+impl LoadBalancer for LeastConnections {
+    fn next(&mut self) -> &mut Server {
+        let mut min = usize::MAX;
+        let mut index = 0;
+        for (i, server) in self.servers.iter().enumerate() {
+            if server.get_clients().len() < min {
+                min = server.get_clients().len();
+                index = i;
+            }
+        }
+        &mut self.servers[index]
+    }
+}

demo/server/rust-example/src/api/loadbalancers/leastconnectionsroundrobin.rs

@@ -0,0 +1,38 @@
+// Least Connection Rounded Robin Load Balancer
+//  This load balancer selects the server with the least number of active connections.
+//  This is done by iterating through the servers, finding the average number of clients and round robin selecting the servers below the average.
+
+use crate::api::{loadbalancer::LoadBalancer, server::Server};
+
+pub struct LeastConnectionsRoundRobin {
+    servers: Vec<Server>,
+    current: i32,
+}
+
+impl LoadBalancer for LeastConnectionsRoundRobin {
+    fn next(&mut self) -> &mut Server {
+        let mut average: i32 = 0;
+        {
+            let mut total_clients: i32 = 0;
+            for server in self.servers.iter() {
+                let clients = server.get_clients().len() as i32;
+                total_clients += clients;
+                average += clients;
+            }
+            
+            average /= total_clients;
+        }
+
+        let mut index = 0;
+        for _ in 0..self.servers.len() {
+            self.current = (self.current + 1) % self.servers.len() as i32;
+            let clients = self.servers[self.current as usize].get_clients().len() as i32;
+            if clients < average {
+                index = self.current;
+                break;
+            }
+        }
+
+        &mut self.servers[index as usize]
+    }
+}

demo/server/rust-example/src/api/loadbalancers/mod.rs

@@ -0,0 +1,3 @@
+pub mod roundrobin;
+pub mod leastconnections;
+pub mod leastconnectionsroundrobin;

demo/server/rust-example/src/api/loadbalancers/roundrobin.rs

@@ -0,0 +1,13 @@
+use crate::api::loadbalancer::LoadBalancer;
+
+pub struct RoundRobin {
+    servers: Vec<super::super::server::Server>,
+    current: i32,
+}
+
+impl LoadBalancer for RoundRobin {
+    fn next(&mut self) -> &mut crate::api::server::Server {
+        self.current = (self.current + 1) % self.servers.len() as i32;
+        &mut self.servers[self.current as usize]
+    }
+}

demo/server/rust-example/src/api/mod.rs

@@ -0,0 +1,4 @@
+pub mod loadbalancer;
+pub mod loadbalancers;
+pub mod server;
+pub mod client;

demo/server/rust-example/src/api/server.rs

@@ -0,0 +1,41 @@
+pub struct Server
+{
+    id: i64,
+    clients: Vec<super::client::Client>,
+}
+
+impl Server
+{
+    pub fn new(id: i64) -> Server
+    {
+        Server
+        {
+            id,
+            clients: Vec::new(),
+        }
+    }
+
+    pub fn get_id(&self) -> i64
+    {
+        self.id
+    }
+
+    pub fn add_client(&mut self, client: super::client::Client)
+    {
+        self.clients.push(client);
+    }
+
+    pub fn get_clients(&self) -> &Vec<super::client::Client>
+    {
+        &self.clients
+    }
+    pub fn remove_client(&mut self, client_id: i64)
+    {
+        self.clients.retain(|client| client.get_id() != client_id);
+    }
+
+    pub fn get_client(&self, client_id: i64) -> Option<&super::client::Client>
+    {
+        self.clients.iter().find(|client| client.get_id() == client_id)
+    }
+}

demo/server/rust-example/src/main.rs

@@ -1,3 +1,79 @@
-fn main() {
-    println!("Hello, world!");
+//! Run with
+//!
+//! ```not_rust
+//! cargo run -p rust-example
+//! ```
+
+use std::net::SocketAddr;
+
+use axum::{
+    extract::{connect_info::ConnectInfo, ws::WebSocketUpgrade}, 
+    response::{Html, IntoResponse}, 
+    routing::get, 
+    Router
+};
+use axum_extra::TypedHeader;
+use headers::UserAgent;
+use tower_http::trace::TraceLayer;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
+
+mod api;
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| format!("{}=debug,tower_http=debug,axum::rejection=trace", env!("CARGO_CRATE_NAME")).into()),
+        )
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+    
+    tracing::info!("Starting server...");
+
+    // build our application with a route
+    let app = Router::new()
+        .route("/", get(handler))
+        .route("/ws", get(ws_handler))
+        .layer(TraceLayer::new_for_http());
+
+    // run it
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000")
+        .await
+        .unwrap();
+
+    tracing::info!("listening on {} - http://localhost:{}/", listener.local_addr().unwrap(), listener.local_addr().unwrap().port());
+    axum::serve(
+        listener, 
+        app.into_make_service_with_connect_info::<SocketAddr>()
+    ).await.unwrap();
 }
+
+async fn ws_handler(
+    ws: WebSocketUpgrade,
+    user_agent: Option<TypedHeader<UserAgent>>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>
+) -> impl IntoResponse {
+    let user_agent = if let Some(TypedHeader(user_agent)) = user_agent {
+        user_agent.to_string()
+    } else {
+        "unknown".to_string()
+    };
+
+    tracing::debug!("WebSocket connection from {} with user agent: {}", addr, user_agent);
+
+    ws.on_upgrade(move |socket| async move {
+        tracing::debug!("WebSocket connection established, from: {}", addr);
+        // handle the WebSocket connection
+
+        // Wait a second.
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        // Close the socket, we don't need it!
+        socket.close().await.unwrap();
+    })
+}
+
+async fn handler() -> Html<&'static str> {
+    Html("<h1>Hello, World!</h1>")
+}