Add initial multiplayer server impl

Relates to DigitalExtinction#26.

Cargo.toml

@@ -70,6 +70,7 @@ de_behaviour = { path = "crates/behaviour", version = "0.1.0-dev" }
 de_camera = { path = "crates/camera", version = "0.1.0-dev" }
 de_combat = { path = "crates/combat", version = "0.1.0-dev" }
 de_conf = { path = "crates/conf", version = "0.1.0-dev" }
+de_connector = { path = "crates/connector", version = "0.1.0-dev" }
 de_construction = { path = "crates/construction", version = "0.1.0-dev" }
 de_controller = { path = "crates/controller", version = "0.1.0-dev" }
 de_core = { path = "crates/core", version = "0.1.0-dev" }
@@ -81,6 +82,8 @@ de_lobby_model = { path = "crates/lobby_model", version = "0.1.0-dev" }
 de_map = { path = "crates/map", version = "0.1.0-dev" }
 de_menu = { path = "crates/menu", version = "0.1.0-dev" }
 de_movement = { path = "crates/movement", version = "0.1.0-dev" }
+de_multiplayer = { path = "crates/multiplayer", version = "0.1.0-dev" }
+de_net = { path = "crates/net", version = "0.1.0-dev" }
 de_objects = { path = "crates/objects", version = "0.1.0-dev" }
 de_pathing = { path = "crates/pathing", version = "0.1.0-dev" }
 de_signs = { path = "crates/signs", version = "0.1.0-dev" }
@@ -101,6 +104,7 @@ criterion = "0.4"
 dirs = "4.0.0"
 enum-iterator = "1.4.0"
 enum-map = "2.3.0"
+futures = "0.3.28"
 futures-lite = "1.11"
 glam = "0.23"
 gltf = "1.0"
@@ -119,5 +123,7 @@ sha3 = "0.10.6"
 spade = "2.0.0"
 thiserror = "1.0"
 tinyvec = { version = "1.6.0", features = ["rustc_1_40", "alloc"] }
+tracing = "0.1.37"
+tracing-subscriber = "0.3.16"
 url = { version = "2.3.1", features = ["serde"] }
 urlencoding = "2.1.2"

crates/connector/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "de_connector"
+description = "Digital Extinction multiplayer server."
+
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+repository.workspace = true
+keywords.workspace = true
+homepage.workspace = true
+license.workspace = true
+categories.workspace = true
+
+[lib]
+name = "de_connector_lib" 
+
+[[bin]]
+name = "de_connector" 
+
+[dependencies]
+# DE
+de_net.workspace = true
+
+# Other
+ahash.workspace = true
+async-std.workspace = true
+futures.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true

crates/connector/src/lib.rs

@@ -0,0 +1,40 @@
+use std::net::SocketAddr;
+
+use ahash::AHashSet;
+use async_std::task;
+use de_net::{Network, MAX_DATAGRAM_SIZE};
+use futures::future::try_join_all;
+use tracing::info;
+
+pub fn start() {
+    info!("Starting...");
+    task::block_on(task::spawn(async { main_loop().await }));
+}
+
+async fn main_loop() {
+    let mut clients: AHashSet<SocketAddr> = AHashSet::new();
+
+    // TODO handle result
+    let mut net = Network::bind().await.unwrap();
+    // TODO handle result
+    info!("Listening on port {}", net.port().unwrap());
+
+    let mut buffer = [0u8; MAX_DATAGRAM_SIZE];
+
+    loop {
+        // TODO handle result
+        let (n, source) = net.recv(&mut buffer).await.unwrap();
+        clients.insert(source);
+
+        let send_futures = clients.iter().filter_map(|&target| {
+            if target == source {
+                None
+            } else {
+                Some(net.send(target, &buffer[0..n]))
+            }
+        });
+
+        // TODO handle result
+        try_join_all(send_futures).await.unwrap();
+    }
+}

crates/connector/src/main.rs

@@ -0,0 +1,11 @@
+use de_connector_lib::start;
+use tracing::Level;
+use tracing_subscriber::FmtSubscriber;
+
+fn main() {
+    let subscriber = FmtSubscriber::builder()
+        .with_max_level(Level::TRACE)
+        .finish();
+    tracing::subscriber::set_global_default(subscriber).unwrap();
+    start();
+}

crates/multiplayer/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "de_multiplayer"
+description = "Digital Extinction multiplayer implementation."
+
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+repository.workspace = true
+keywords.workspace = true
+homepage.workspace = true
+license.workspace = true
+categories.workspace = true
+
+[dependencies]
+# DE
+de_net.workspace = true

crates/multiplayer/src/lib.rs

@@ -0,0 +1 @@
+

crates/net/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "de_net"
+description = "Digital Extinction low latency networking."
+
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+repository.workspace = true
+keywords.workspace = true
+homepage.workspace = true
+license.workspace = true
+categories.workspace = true
+
+[dependencies]
+# Other
+async-std.workspace = true
+thiserror.workspace = true

crates/net/src/lib.rs

@@ -0,0 +1,3 @@
+pub use net::{Network, RecvError, SendError, MAX_DATAGRAM_SIZE};
+
+mod net;

crates/net/src/net.rs

@@ -0,0 +1,81 @@
+use std::io;
+
+use async_std::net::{SocketAddr, UdpSocket};
+use thiserror::Error;
+
+/// Maximum size of a UDP datagram which might be sent by this crate.
+///
+/// For the sake of simplicity, this is currently a value smaller than any
+/// widely used MTU.
+pub const MAX_DATAGRAM_SIZE: usize = 512;
+
+/// This struct represents a low level network connection. The connection is
+/// based on UDP and is unreliable and unordered.
+pub struct Network {
+    socket: UdpSocket,
+}
+
+impl Network {
+    /// Create a new IPv4 based connection (socket) an a system assigned port
+    /// number.
+    pub async fn bind() -> io::Result<Self> {
+        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+        let socket = UdpSocket::bind(addr).await?;
+
+        Ok(Self { socket })
+    }
+
+    pub fn port(&self) -> io::Result<u16> {
+        self.socket.local_addr().map(|addr| addr.port())
+    }
+
+    /// Receive a single datagram.
+    ///
+    /// The returned data are guaranteed to be at most [`MAX_DATAGRAM_SIZE`]
+    /// bytes long.
+    pub async fn recv(&mut self, buf: &mut [u8]) -> Result<(usize, SocketAddr), RecvError> {
+        self.socket.recv_from(buf).await.map_err(RecvError::from)
+    }
+
+    /// Send data to a single target.
+    ///
+    /// # Panics
+    ///
+    /// This method panics if `data` have more than [`MAX_DATAGRAM_SIZE`]
+    /// bytes.
+    pub async fn send(&self, target: SocketAddr, data: &[u8]) -> Result<(), SendError> {
+        if data.len() > MAX_DATAGRAM_SIZE {
+            panic!(
+                "Max datagram size is {} got {}.",
+                MAX_DATAGRAM_SIZE,
+                data.len()
+            );
+        }
+
+        let n = self
+            .socket
+            .send_to(data, target)
+            .await
+            .map_err(SendError::from)?;
+
+        if n < data.len() {
+            Err(SendError::PartialSend(n, data.len()))
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[derive(Error, Debug)]
+pub enum RecvError {
+    #[error("an IO error occurred")]
+    Io(#[from] io::Error),
+}
+
+#[derive(Error, Debug)]
+pub enum SendError {
+    #[error("an IO error occurred")]
+    Io(#[from] io::Error),
+    #[error("only {0} of {1} bytes sent")]
+    PartialSend(usize, usize),
+}