Rewrite server

Author: skade

examples/a-chat/server.rs

@@ -3,18 +3,17 @@ use std::{
     sync::Arc,
 };
 
-use futures::{channel::mpsc, select, FutureExt, SinkExt};
+use futures::{select, FutureExt};
 
 use async_std::{
     io::BufReader,
     net::{TcpListener, TcpStream, ToSocketAddrs},
     prelude::*,
     task,
+    sync::{channel, Sender, Receiver},
 };
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
-type Sender<T> = mpsc::UnboundedSender<T>;
-type Receiver<T> = mpsc::UnboundedReceiver<T>;
 
 #[derive(Debug)]
 enum Void {}
@@ -26,7 +25,7 @@ pub(crate) fn main() -> Result<()> {
 async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
     let listener = TcpListener::bind(addr).await?;
 
-    let (broker_sender, broker_receiver) = mpsc::unbounded();
+    let (broker_sender, broker_receiver) = channel(10);
     let broker = task::spawn(broker_loop(broker_receiver));
     let mut incoming = listener.incoming();
     while let Some(stream) = incoming.next().await {
@@ -39,7 +38,7 @@ async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
     Ok(())
 }
 
-async fn connection_loop(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
+async fn connection_loop(broker: Sender<Event>, stream: TcpStream) -> Result<()> {
     let stream = Arc::new(stream);
     let reader = BufReader::new(&*stream);
     let mut lines = reader.lines();
@@ -48,15 +47,14 @@ async fn connection_loop(mut broker: Sender<Event>, stream: TcpStream) -> Result
         None => return Err("peer disconnected immediately".into()),
         Some(line) => line?,
     };
-    let (_shutdown_sender, shutdown_receiver) = mpsc::unbounded::<Void>();
+    let (_shutdown_sender, shutdown_receiver) = channel::<Void>(0);
     broker
         .send(Event::NewPeer {
             name: name.clone(),
             stream: Arc::clone(&stream),
             shutdown: shutdown_receiver,
         })
-        .await
-        .unwrap();
+        .await;
 
     while let Some(line) = lines.next().await {
         let line = line?;
@@ -76,8 +74,7 @@ async fn connection_loop(mut broker: Sender<Event>, stream: TcpStream) -> Result
                 to: dest,
                 msg,
             })
-            .await
-            .unwrap();
+            .await;
     }
 
     Ok(())
@@ -115,23 +112,23 @@ enum Event {
         from: String,
         to: Vec<String>,
         msg: String,
-    },
+    }
 }
 
 async fn broker_loop(mut events: Receiver<Event>) {
     let (disconnect_sender, mut disconnect_receiver) =
-        mpsc::unbounded::<(String, Receiver<String>)>();
+        channel::<(String, Receiver<String>)>(2);
     let mut peers: HashMap<String, Sender<String>> = HashMap::new();
 
     loop {
+
         let event = select! {
-            event = events.next().fuse() => match event {
+            event = events.next() => match event {
                 None => break,
                 Some(event) => event,
             },
-            disconnect = disconnect_receiver.next().fuse() => {
+            disconnect = disconnect_receiver.next() => {
                 let (name, _pending_messages) = disconnect.unwrap();
-                assert!(peers.remove(&name).is_some());
                 continue;
             },
         };
@@ -140,7 +137,7 @@ async fn broker_loop(mut events: Receiver<Event>) {
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
                         let msg = format!("from {}: {}\n", from, msg);
-                        peer.send(msg).await.unwrap();
+                        peer.send(msg).await;
                     }
                 }
             }
@@ -151,20 +148,24 @@ async fn broker_loop(mut events: Receiver<Event>) {
             } => match peers.entry(name.clone()) {
                 Entry::Occupied(..) => (),
                 Entry::Vacant(entry) => {
-                    let (client_sender, mut client_receiver) = mpsc::unbounded();
+                    let (client_sender, mut client_receiver) = channel(10);
                     entry.insert(client_sender);
-                    let mut disconnect_sender = disconnect_sender.clone();
+                    let  disconnect_sender = disconnect_sender.clone();
                     spawn_and_log_error(async move {
                         let res =
                             connection_writer_loop(&mut client_receiver, stream, shutdown).await;
                         disconnect_sender
                             .send((name, client_receiver))
-                            .await
-                            .unwrap();
+                            .await;
                         res
                     });
                 }
             },
+            Event::Disconnect {
+                name
+            } => {
+                assert!(peers.remove(&name).is_some());
+            }
         }
     }
     drop(peers);