feather-rs
diff --git a/‎Cargo.lock‎
Lines changed: 1 addition & 0 deletions b/‎Cargo.lock‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/server/Cargo.toml‎
Lines changed: 1 addition & 0 deletions b/‎crates/server/Cargo.toml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/server/src/connection_worker.rs‎
Lines changed: 42 additions & 6 deletions b/‎crates/server/src/connection_worker.rs‎
Lines changed: 42 additions & 6 deletions
diff --git a/‎crates/server/src/initial_handler.rs‎
Lines changed: 2 additions & 2 deletions b/‎crates/server/src/initial_handler.rs‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎crates/server/src/lib.rs‎
Lines changed: 12 additions & 1 deletion b/‎crates/server/src/lib.rs‎
Lines changed: 12 additions & 1 deletion
diff --git a/‎crates/server/src/listener.rs‎
Lines changed: 8 additions & 1 deletion b/‎crates/server/src/listener.rs‎
Lines changed: 8 additions & 1 deletion
diff --git a/‎crates/server/src/player_count.rs‎
Lines changed: 105 additions & 0 deletions b/‎crates/server/src/player_count.rs‎
Lines changed: 105 additions & 0 deletions
@@ -22,6 +22,7 @@ plugin-host = { path = "../plugin-host", package = "feather-plugin-host" }
 
 ahash = "0.7"
 anyhow = "1"
+crossbeam-utils = "0.8"
 tokio = { version = "1", features = ["full"] }
 flume = "0.10"
 futures-lite = "1"
 
@@ -1,10 +1,12 @@
 use std::{fmt::Debug, io, net::SocketAddr, sync::Arc, time::Duration};
 
+use base::Text;
 use flume::{Receiver, Sender};
 use futures_lite::FutureExt;
 use io::ErrorKind;
 use protocol::{
-    codec::CryptKey, ClientPlayPacket, MinecraftCodec, Readable, ServerPlayPacket, Writeable,
+    codec::CryptKey, packets::server::Disconnect, ClientPlayPacket, MinecraftCodec, Readable,
+    ServerPlayPacket, Writeable,
 };
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
@@ -18,6 +20,7 @@ use tokio::{
 use crate::{
     initial_handler::{InitialHandling, NewPlayer},
     options::Options,
+    player_count::PlayerCount,
 };
 
 /// Tokio task which handles a connection and processes
@@ -32,6 +35,7 @@ pub struct Worker {
     reader: Reader,
     writer: Writer,
     options: Arc<Options>,
+    player_count: PlayerCount,
     packets_to_send_tx: Sender<ServerPlayPacket>,
     received_packets_rx: Receiver<ClientPlayPacket>,
     new_players: Sender<NewPlayer>,
@@ -42,6 +46,7 @@ impl Worker {
         stream: TcpStream,
         _addr: SocketAddr,
         options: Arc<Options>,
+        player_count: PlayerCount,
         new_players: Sender<NewPlayer>,
     ) -> Self {
         let (reader, writer) = stream.into_split();
@@ -55,6 +60,7 @@ impl Worker {
             reader,
             writer,
             options,
+            player_count,
             packets_to_send_tx,
             received_packets_rx,
             new_players,
@@ -75,12 +81,22 @@ impl Worker {
         }
     }
 
-    async fn proceed(self, result: InitialHandling) {
+    async fn proceed(mut self, result: InitialHandling) {
         match result {
             InitialHandling::Disconnect => (),
             InitialHandling::Join(new_player) => {
+                if self.player_count.try_add_player().is_err() {
+                    self.write(ServerPlayPacket::Disconnect(Disconnect {
+                        reason: Text::from("The server is full!").to_string(),
+                    }))
+                    .await
+                    .ok();
+                    return;
+                }
+
+                let username = new_player.username.clone();
                 let _ = self.new_players.send_async(new_player).await;
-                self.split();
+                self.split(username);
             }
         }
     }
@@ -89,6 +105,10 @@ impl Worker {
         &self.options
     }
 
+    pub fn player_count(&self) -> u32 {
+        self.player_count.get()
+    }
+
     #[allow(unused)]
     pub fn enable_compression(&mut self, threshold: usize) {
         self.reader.codec.enable_compression(threshold);
@@ -108,16 +128,23 @@ impl Worker {
         self.writer.write(packet).await
     }
 
-    pub fn split(self) {
-        let Self { reader, writer, .. } = self;
+    pub fn split(self, username: String) {
+        let Self {
+            reader,
+            writer,
+            player_count,
+            ..
+        } = self;
         let reader = tokio::task::spawn(async move { reader.run().await });
         let writer = tokio::task::spawn(async move { writer.run().await });
 
         tokio::task::spawn(async move {
             let result = reader.race(writer).await.expect("task panicked");
             if let Err(e) = result {
-                log::error!("Connection lost: {:?}", e);
+                let message = disconnected_message(e);
+                log::debug!("{} lost connection: {}", username, message);
             }
+            player_count.remove_player();
         });
     }
 
@@ -208,3 +235,12 @@ impl Writer {
         Ok(())
     }
 }
+
+fn disconnected_message(e: anyhow::Error) -> String {
+    if let Some(io_error) = e.downcast_ref::<io::Error>() {
+        if io_error.kind() == ErrorKind::UnexpectedEof {
+            return "disconnected".to_owned();
+        }
+    }
+    format!("{:?}", e)
+}
@@ -78,7 +78,7 @@ struct Version {
 #[derive(Debug, Serialize)]
 struct Players {
     max: u32,
-    online: usize,
+    online: u32,
 }
 
 async fn handle_status(worker: &mut Worker) -> anyhow::Result<InitialHandling> {
@@ -91,7 +91,7 @@ async fn handle_status(worker: &mut Worker) -> anyhow::Result<InitialHandling> {
         },
         players: Players {
             max: worker.options().max_players,
-            online: 0, // TODO (1.16): player count
+            online: worker.player_count(),
         },
         description: Text::from(worker.options().motd.clone()),
         favicon: worker
 
@@ -21,11 +21,13 @@ mod listener;
 mod network_id_registry;
 mod options;
 mod packet_handlers;
+mod player_count;
 mod systems;
 
 pub use client::{Client, ClientId, Clients};
 pub use network_id_registry::NetworkId;
 pub use options::Options;
+use player_count::PlayerCount;
 use systems::view::WaitingChunks;
 
 /// A Minecraft server.
@@ -44,6 +46,8 @@ pub struct Server {
     chunk_subscriptions: ChunkSubscriptions,
 
     last_keepalive_time: Instant,
+
+    player_count: PlayerCount,
 }
 
 impl Server {
@@ -52,9 +56,10 @@ impl Server {
     /// Must be called within the context of a Tokio runtime.
     pub async fn bind(options: Options) -> anyhow::Result<Self> {
         let options = Arc::new(options);
+        let player_count = PlayerCount::new(options.max_players);
 
         let (new_players_tx, new_players) = flume::bounded(4);
-        Listener::start(Arc::clone(&options), new_players_tx).await?;
+        Listener::start(Arc::clone(&options), player_count.clone(), new_players_tx).await?;
 
         log::info!(
             "Server is listening on {}:{}",
@@ -69,6 +74,7 @@ impl Server {
             waiting_chunks: WaitingChunks::default(),
             chunk_subscriptions: ChunkSubscriptions::default(),
             last_keepalive_time: Instant::now(),
+            player_count,
         })
     }
 
@@ -78,6 +84,11 @@ impl Server {
         systems::register(self, game, systems);
         game.add_entity_spawn_callback(entities::add_entity_components);
     }
+
+    /// Gets the number of online players.
+    pub fn player_count(&self) -> u32 {
+        self.player_count.get()
+    }
 }
 
 /// Low-level functions, mostly used internally.
 
@@ -4,18 +4,23 @@ use anyhow::Context;
 use flume::Sender;
 use tokio::net::{TcpListener, TcpStream};
 
-use crate::{connection_worker::Worker, initial_handler::NewPlayer, options::Options};
+use crate::{
+    connection_worker::Worker, initial_handler::NewPlayer, options::Options,
+    player_count::PlayerCount,
+};
 
 /// Listens for and accepts incoming connections.
 pub struct Listener {
     listener: TcpListener,
     options: Arc<Options>,
+    player_count: PlayerCount,
     new_players: Sender<NewPlayer>,
 }
 
 impl Listener {
     pub async fn start(
         options: Arc<Options>,
+        player_count: PlayerCount,
         new_players: Sender<NewPlayer>,
     ) -> anyhow::Result<()> {
         let listener = TcpListener::bind(format!("{}:{}", options.bind_address, options.port))
@@ -25,6 +30,7 @@ impl Listener {
         let listener = Listener {
             listener,
             options,
+            player_count,
             new_players,
         };
         tokio::task::spawn(async move {
@@ -47,6 +53,7 @@ impl Listener {
             stream,
             addr,
             Arc::clone(&self.options),
+            self.player_count.clone(),
             self.new_players.clone(),
         );
         worker.start();
 
@@ -0,0 +1,105 @@
+use std::sync::{
+    atomic::{AtomicU32, Ordering},
+    Arc,
+};
+
+#[derive(Debug)]
+pub struct MaxPlayersReached;
+
+/// Maintains the server player count.
+///
+/// Can be cloned to create a new handle.
+#[derive(Clone)]
+pub struct PlayerCount {
+    inner: Arc<Inner>,
+}
+
+impl PlayerCount {
+    pub fn new(max_players: u32) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                count: AtomicU32::new(0),
+                max_players,
+            }),
+        }
+    }
+
+    pub fn try_add_player(&self) -> Result<(), MaxPlayersReached> {
+        loop {
+            let current_count = self.inner.count.load(Ordering::SeqCst);
+            let new_count = current_count + 1;
+            if new_count > self.inner.max_players {
+                return Err(MaxPlayersReached);
+            }
+
+            if self
+                .inner
+                .count
+                .compare_exchange(current_count, new_count, Ordering::SeqCst, Ordering::SeqCst)
+                .is_ok()
+            {
+                return Ok(());
+            }
+        }
+    }
+
+    pub fn remove_player(&self) {
+        self.inner.count.fetch_sub(1, Ordering::SeqCst);
+    }
+
+    pub fn get(&self) -> u32 {
+        self.inner.count.load(Ordering::Acquire)
+    }
+}
+
+struct Inner {
+    count: AtomicU32,
+    max_players: u32,
+}
+
+#[cfg(test)]
+mod tests {
+    use crossbeam_utils::thread;
+
+    use super::*;
+
+    #[test]
+    fn try_add() {
+        let count = PlayerCount::new(1);
+        assert_eq!(count.get(), 0);
+        count.try_add_player().unwrap();
+        assert_eq!(count.get(), 1);
+
+        for _ in 0..10 {
+            count.try_add_player().unwrap_err();
+            assert_eq!(count.get(), 1);
+        }
+    }
+
+    #[test]
+    fn no_race_conditions() {
+        let threads = 8;
+        let players_per_thread = 100000;
+        let max_players = threads * players_per_thread / 2;
+
+        let count = PlayerCount::new(max_players);
+
+        let num_added = AtomicU32::new(0);
+
+        thread::scope(|s| {
+            for _ in 0..threads {
+                s.spawn(|_| {
+                    for _ in 0..players_per_thread {
+                        if count.try_add_player().is_ok() {
+                            num_added.fetch_add(1, Ordering::SeqCst);
+                        }
+                    }
+                });
+            }
+        })
+        .unwrap();
+
+        assert_eq!(num_added.load(Ordering::SeqCst), max_players);
+        assert_eq!(count.get(), max_players);
+    }
+}