First publication.

Author: jimblandy

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "async-chat"
+version = "0.1.0"
+authors = ["Jim Blandy <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+async-std = { version = "1.7", features = ["unstable"] }
+tokio = { version = "0.3", features = ["sync"] }
+serde = { version = "1.0", features = ["derive", "rc"] }
+serde_json = "1.0"

rustfmt.toml

@@ -0,0 +1,2 @@
+# Lines wider than 80 columns don't fit in the book so well.
+max_width = 80

src/bin/client.rs

@@ -0,0 +1,115 @@
+//# send-commands
+use async_std::prelude::*;
+use async_chat::utils::{self, ChatResult};
+use async_std::io;
+use async_std::net;
+
+async fn send_commands(mut to_server: net::TcpStream) -> ChatResult<()> {
+    println!("Commands:\n\
+              join GROUP\n\
+              post GROUP MESSAGE...\n\
+              Type Control-D (on Unix) or Control-Z (on Windows) \
+              to close the connection.");
+
+    let mut command_lines = io::BufReader::new(io::stdin()).lines();
+    while let Some(command) = command_lines.next().await {
+        let command = command?;
+        let request = match parse_command(&command) {
+            Some(request) => request,
+            None => continue,
+        };
+
+        utils::send_as_json(&mut to_server, &request).await?;
+        to_server.flush().await?;
+    }
+
+    Ok(())
+}
+//#end
+
+//# client-handle-replies
+use async_chat::FromServer;
+
+async fn handle_replies(from_server: net::TcpStream) -> ChatResult<()> {
+    let buffered = io::BufReader::new(from_server);
+    let mut reply_stream = utils::receive_as_json(buffered);
+
+    while let Some(reply) = reply_stream.next().await {
+        match reply? {
+            FromServer::Message { group_name, message } => {
+                println!("message posted to {}: {}", group_name, message);
+            }
+            FromServer::Error(message) => {
+                println!("error from server: {}", message);
+            }
+        }
+    }
+
+    Ok(())
+}
+//# end
+
+//# client-main
+use async_std::task;
+
+fn main() -> ChatResult<()> {
+    let address = std::env::args().nth(1)
+        .expect("Usage: client ADDRESS:PORT");
+
+    task::block_on(async {
+        let socket = net::TcpStream::connect(address).await?;
+        socket.set_nodelay(true)?;
+
+        let to_server = send_commands(socket.clone());
+        let from_server = handle_replies(socket);
+
+        from_server.race(to_server).await?;
+
+        Ok(())
+    })
+}
+//# end
+
+use async_chat::FromClient;
+use std::sync::Arc;
+
+/// Parse a line (presumably read from the standard input) as a `Request`.
+fn parse_command(line: &str) -> Option<FromClient> {
+    let (command, rest) = get_next_token(line)?;
+    if command == "post" {
+        let (group, rest) = get_next_token(rest)?;
+        let message = rest.trim_start().to_string();
+        return Some(FromClient::Post {
+            group_name: Arc::new(group.to_string()),
+            message: Arc::new(message),
+        });
+    } else if command == "join" {
+        let (group, rest) = get_next_token(rest)?;
+        if !rest.trim_start().is_empty() {
+            return None;
+        }
+        return Some(FromClient::Join {
+            group_name: Arc::new(group.to_string()),
+        });
+    } else {
+        eprintln!("Unrecognized command: {:?}", line);
+        return None;
+    }
+}
+
+/// Given a string `input`, return `Some((token, rest))`, where `token` is the
+/// first run of non-whitespace characters in `input`, and `rest` is the rest of
+/// the string. If the string contains no non-whitespace characters, return
+/// `None`.
+fn get_next_token(mut input: &str) -> Option<(&str, &str)> {
+    input = input.trim_start();
+
+    if input.is_empty() {
+        return None;
+    }
+
+    match input.find(char::is_whitespace) {
+        Some(space) => Some((&input[0..space], &input[space..])),
+        None => Some((input, "")),
+    }
+}

src/bin/server/connection.rs

@@ -0,0 +1,70 @@
+/// Handle a single client's connection.
+
+//# chat-serve
+use async_chat::{FromClient, FromServer};
+use async_chat::utils::{self, ChatResult};
+use async_std::prelude::*;
+use async_std::io::BufReader;
+use async_std::net::TcpStream;
+use async_std::sync::Arc;
+
+use crate::group_table::GroupTable;
+
+pub async fn serve(socket: TcpStream, groups: Arc<GroupTable>)
+                   -> ChatResult<()>
+{
+    let outbound = Arc::new(Outbound::new(socket.clone()));
+
+    let buffered = BufReader::new(socket);
+    let mut from_client = utils::receive_as_json(buffered);
+    while let Some(request) = from_client.next().await {
+        let request = request?;
+
+        let result = match request {
+            FromClient::Join { group_name } => {
+                let group = groups.get_or_create(group_name);
+                group.join(outbound.clone());
+                Ok(())
+            }
+
+            FromClient::Post { group_name, message } => {
+                match groups.get(&group_name) {
+                    Some(group) => {
+                        group.post(message);
+                        Ok(())
+                    }
+                    None => {
+                        Err(format!("Group '{}' does not exist", group_name))
+                    }
+                }
+            }
+        };
+
+        if let Err(message) = result {
+            let report = FromServer::Error(message);
+            outbound.send(report).await?;
+        }
+    }
+
+    Ok(())
+}
+//# end
+
+//# chat-Outbound
+use async_std::sync::Mutex;
+
+pub struct Outbound(Mutex<TcpStream>);
+
+impl Outbound {
+    pub fn new(to_client: TcpStream) -> Outbound {
+        Outbound(Mutex::new(to_client))
+    }
+
+    pub async fn send(&self, packet: FromServer) -> ChatResult<()> {
+        let mut guard = self.0.lock().await;
+        utils::send_as_json(&mut *guard, &packet).await?;
+        guard.flush().await?;
+        Ok(())
+    }
+}
+//# end

src/bin/server/group.rs

@@ -0,0 +1,65 @@
+//! A chat group.
+
+//# group-type
+use async_std::task;
+use crate::connection::Outbound;
+use std::sync::Arc;
+use tokio::sync::broadcast;
+
+pub struct Group {
+    name: Arc<String>,
+    sender: broadcast::Sender<Arc<String>>
+}
+
+impl Group {
+    pub fn new(name: Arc<String>) -> Group {
+        let (sender, _receiver) = broadcast::channel(1000);
+        Group { name, sender }
+    }
+
+    pub fn join(&self, outbound: Arc<Outbound>) {
+        let receiver = self.sender.subscribe();
+
+        task::spawn(handle_subscriber(self.name.clone(),
+                                      receiver,
+                                      outbound));
+    }
+
+    pub fn post(&self, message: Arc<String>) {
+        // This only returns an error when there are no subscribers. A
+        // connection's outgoing side can exit, dropping its subscription,
+        // slightly before its incoming side, which may end up trying to send a
+        // message to an empty group.
+        let _ignored = self.sender.send(message);
+    }
+}
+//# end
+
+//# handle_subscriber
+use async_chat::FromServer;
+use tokio::sync::broadcast::error::RecvError;
+
+async fn handle_subscriber(group_name: Arc<String>,
+                           mut receiver: broadcast::Receiver<Arc<String>>,
+                           outbound: Arc<Outbound>)
+{
+    loop {
+        let packet = match receiver.recv().await {
+            Ok(message) => FromServer::Message {
+                group_name: group_name.clone(),
+                message: message.clone(),
+            },
+
+            Err(RecvError::Lagged(n)) => FromServer::Error(
+                format!("Dropped {} messages from {}.", n, group_name)
+            ),
+
+            Err(RecvError::Closed) => break,
+        };
+
+        if outbound.send(packet).await.is_err() {
+            break;
+        }
+    }
+}
+//# end

src/bin/server/group_table.rs

@@ -0,0 +1,29 @@
+//# group-table
+use crate::group::Group;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+pub struct GroupTable(Mutex<HashMap<Arc<String>, Arc<Group>>>);
+
+impl GroupTable {
+    pub fn new() -> GroupTable {
+        GroupTable(Mutex::new(HashMap::new()))
+    }
+
+    pub fn get(&self, name: &String) -> Option<Arc<Group>> {
+        self.0.lock()
+            .unwrap()
+            .get(name)
+            .cloned()
+    }
+
+    pub fn get_or_create(&self, name: Arc<String>) -> Arc<Group> {
+        self.0.lock()
+            .unwrap()
+            .entry(name.clone())
+            .or_insert_with(|| Arc::new(Group::new(name)))
+            .clone()
+    }
+}
+//# end
+

src/bin/server/main.rs

@@ -0,0 +1,33 @@
+//! Asynchronous chat server.
+
+//# server-main
+use async_std::prelude::*;
+use async_std::{net, task};
+use async_chat::utils::ChatResult;
+use std::sync::Arc;
+
+mod connection;
+mod group;
+mod group_table;
+
+fn main() -> ChatResult<()> {
+    let address = std::env::args().nth(1).expect("Usage: server ADDRESS");
+
+    let groups = Arc::new(group_table::GroupTable::new());
+
+    task::block_on(async {
+        let listener = net::TcpListener::bind(address).await?;
+
+        let mut new_connections = listener.incoming();
+        loop {
+            let socket = new_connections.next().await.unwrap()?;
+            let groups = groups.clone();
+            task::spawn(async {
+                if let Err(error) = connection::serve(socket, groups).await {
+                    eprintln!("Error: {}", error);
+                }
+            });
+        }
+    })
+}
+//# end

src/lib.rs

@@ -0,0 +1,24 @@
+pub mod utils;
+
+//# request-and-reply
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+#[derive(Debug, Deserialize, Serialize, PartialEq)]
+pub enum FromClient {
+    Join { group_name: Arc<String> },
+    Post {
+        group_name: Arc<String>,
+        message: Arc<String>,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize, PartialEq)]
+pub enum FromServer {
+    Message {
+        group_name: Arc<String>,
+        message: Arc<String>,
+    },
+    Error(String),
+}
+//# end