All Together

此时，我们只需要开始 broker，就可获得功能齐全的聊天室（在愉快的情况下！）：

  1. # extern crate async_std;
  2. # extern crate futures;
  3. use async_std::{
  4.     io::{self, BufReader},
  5.     net::{TcpListener, TcpStream, ToSocketAddrs},
  6.     prelude::*,
  7.     task,
  8. };
  9. use futures::channel::mpsc;
  10. use futures::SinkExt;
  11. use std::{
  12.     collections::hash_map::{HashMap, Entry},
  13.     sync::Arc,
  14. };
  16. type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
  17. type Sender<T> = mpsc::UnboundedSender<T>;
  18. type Receiver<T> = mpsc::UnboundedReceiver<T>;
  20. // main
  21. fn run() -> Result<()> {
  22.     task::block_on(accept_loop("127.0.0.1:8080"))
  23. }
  25. fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
  26. where
  27.     F: Future<Output = Result<()>> + Send + 'static,
  28. {
  29.     task::spawn(async move {
  30.         if let Err(e) = fut.await {
  31.             eprintln!("{}", e)
  32.         }
  33.     })
  34. }
  36. async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
  37.     let listener = TcpListener::bind(addr).await?;
  39.     let (broker_sender, broker_receiver) = mpsc::unbounded(); // 1
  40.     let _broker_handle = task::spawn(broker_loop(broker_receiver));
  41.     let mut incoming = listener.incoming();
  42.     while let Some(stream) = incoming.next().await {
  43.         let stream = stream?;
  44.         println!("Accepting from: {}", stream.peer_addr()?);
  45.         spawn_and_log_error(connection_loop(broker_sender.clone(), stream));
  46.     }
  47.     Ok(())
  48. }
  50. async fn connection_loop(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
  51.     let stream = Arc::new(stream); // 2
  52.     let reader = BufReader::new(&*stream);
  53.     let mut lines = reader.lines();
  55.     let name = match lines.next().await {
  56.         None => Err("peer disconnected immediately")?,
  57.         Some(line) => line?,
  58.     };
  59.     broker.send(Event::NewPeer { name: name.clone(), stream: Arc::clone(&stream) }).await // 3
  60.         .unwrap();
  62.     while let Some(line) = lines.next().await {
  63.         let line = line?;
  64.         let (dest, msg) = match line.find(':') {
  65.             None => continue,
  66.             Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
  67.         };
  68.         let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
  69.         let msg: String = msg.to_string();
  71.         broker.send(Event::Message { // 4
  72.             from: name.clone(),
  73.             to: dest,
  74.             msg,
  75.         }).await.unwrap();
  76.     }
  77.     Ok(())
  78. }
  80. async fn connection_writer_loop(
  81.     mut messages: Receiver<String>,
  82.     stream: Arc<TcpStream>,
  83. ) -> Result<()> {
  84.     let mut stream = &*stream;
  85.     while let Some(msg) = messages.next().await {
  86.         stream.write_all(msg.as_bytes()).await?;
  87.     }
  88.     Ok(())
  89. }
  91. #[derive(Debug)]
  92. enum Event {
  93.     NewPeer {
  94.         name: String,
  95.         stream: Arc<TcpStream>,
  96.     },
  97.     Message {
  98.         from: String,
  99.         to: Vec<String>,
  100.         msg: String,
  101.     },
  102. }
  104. async fn broker_loop(mut events: Receiver<Event>) -> Result<()> {
  105.     let mut peers: HashMap<String, Sender<String>> = HashMap::new();
  107.     while let Some(event) = events.next().await {
  108.         match event {
  109.             Event::Message { from, to, msg } => {
  110.                 for addr in to {
  111.                     if let Some(peer) = peers.get_mut(&addr) {
  112.                         let msg = format!("from {}: {}\n", from, msg);
  113.                         peer.send(msg).await?
  114.                     }
  115.                 }
  116.             }
  117.             Event::NewPeer { name, stream} => {
  118.                 match peers.entry(name) {
  119.                     Entry::Occupied(..) => (),
  120.                     Entry::Vacant(entry) => {
  121.                         let (client_sender, client_receiver) = mpsc::unbounded();
  122.                         entry.insert(client_sender); // 4
  123.                         spawn_and_log_error(connection_writer_loop(client_receiver, stream)); // 5
  124.                     }
  125.                 }
  126.             }
  127.         }
  128.     }
  129.     Ok(())
  130. }
  1. 在accept_loop里面，我们创建了 broker 的 channel，和task。
  2. 在connection_loop里面，我们需要将TcpStream包进一个Arc，以便能够与connection_writer_loop共享它。
  3. 登录后，我们会通知 broker。注意我们.unwrap on send： broker 应该比所有客户都长命，如果不是这种情况， broker 应该 panics，所以这个 panics 应该是致命的。
  4. 同样，我们将已解析的消息转发给 broker，并假设该消息仍处于活动状态。