示例：聊天服务器

示例：聊天服务器

我们将使用到目前为止已经涵盖的内容来构建聊天服务器。这是一个非平凡的Tokio服务器应用程序。

服务器将使用基于行的协议。行以\ r \ n结束。这与telnet兼容，因此我们只使用telnet作为客户端。当客户端连接时，它必须通过发送包含其“缺口”的行来标识自己（即，用于在其 Peer中标识客户端的某个名称）。

识别出客户端后，所有发送的行都以[nick]:为前缀,并广播给所有其他连接的客户端。

完整的代码可以在这里找到。请注意，Tokio提供了一些尚未涵盖的额外抽象，这些抽象将使聊天服务器能够用更少的代码编写。

首先，生成一个新的箱子。

1. $ cargo new --bin line-chat
2. cd line-chat

接下来，添加必要的依赖项：

1. [dependencies]
2. tokio = "0.1"
3. tokio-io = "0.1"
4. futures = "0.1"
5. bytes = "0.4"

1. extern crate tokio;
2. #[macro_use]
3. extern crate futures;
4. extern crate bytes;

6. use tokio::io;
7. use tokio::net::{TcpListener, TcpStream};
8. use tokio::prelude::*;
9. use futures::sync::mpsc;
10. use futures::future::{self, Either};
11. use bytes::{BytesMut, Bytes, BufMut};

13. use std::collections::HashMap;
14. use std::net::SocketAddr;
15. use std::sync::{Arc, Mutex};

18. type Tx = mpsc::UnboundedSender<Bytes>;

21. type Rx = mpsc::UnboundedReceiver<Bytes>;

现在，我们为服务器设置必要的结构。这些步骤与Hello World中使用的步骤相同！例：

将TcpListener绑定到本地端口。
定义接受入站连接并处理它们的任务。
启动Tokio运行时
产生服务器任务。同样，在执行程序上生成服务器任务之前，实际上不会发生任何工作。

1. fn main() {
2. let addr = "127.0.0.1:6142".parse().unwrap();
3. let listener = TcpListener::bind(&addr).unwrap();

5. let server = listener.incoming().for_each(move |socket| {

7. Ok(())
8. })
9. .map_err(|err| {

11. println!("accept error = {:?}", err);
12. });

14. println!("server running on localhost:6142");

24. tokio::run(server);
25. }

Chat State

聊天服务器要求从一个客户端接收的消息被广播到所有其他连接的客户端。这将使用通过mpsc通道传递的消息来完成。

每个客户端套接字都将由任务管理。每个任务都有一个关联的mpsc通道，用于接收来自其他客户端的消息。所有这些通道的发送一半存储在Rc单元中以使它们可访问。

在这个例子中，我们将使用无界通道。理想情况下，渠道永远不应该是无限制的，但在这种情况下处理背压有点棘手。我们将把通道限制在后面专门用于处理背压的部分。

以下是共享状态的定义方式（上面已完成Tx类型别名）：

1. struct Shared {
2. peers: HashMap<SocketAddr, Tx>,
3. }

然后，在main函数的最顶部，创建状态实例。此状态实例将移动到接受传入连接的任务中。

1. let state = Arc::new(Mutex::new(Shared::new()));

现在我们可以处理传入的连接。服务器任务更新为：

1. listener.incoming().for_each(move |socket| {
2. process(socket, state.clone());
3. Ok(())
4. })

服务器任务将所有套接字以及服务器状态的克隆传递给进程函数。我们来定义那个功能。它将具有这样的结构：

1. fn process(socket: TcpStream, state: Arc<Mutex<Shared>>) {

3. let task = unimplemented!();

6. tokio::spawn(task);
7. }

对tokio :: spawn的调用将在当前的Tokio运行时生成一个新任务。所有工作线程都保留对存储在线程局部变量中的当前运行时的引用。注意，尝试从Tokio运行时外部调用tokio :: spawn将导致恐慌。

所有连接处理逻辑必须能够理解协议。该协议是基于行的，由\ r \ n终止。它不是在字节流级别工作，而是更容易在帧级工作，即使用表示原子消息的值。

我们实现了一个包含套接字的编解码器，并公开了一个采用和消耗行的API。

线性编解码器

对于采用字节流类型（AsyncRead + AsyncWrite）并在帧级别公开读写API的类型，编解码器是一个松散术语。 tokio-io crate为编写编解码器提供了额外的帮助，在这个例子中，我们将手动完成。

Lines编解码器定义如下：

1. struct Lines {
2. socket: TcpStream,
3. rd: BytesMut,
4. wr: BytesMut,
5. }

7. impl Lines {

9. fn new(socket: TcpStream) -> Self {
10. Lines {
11. socket,
12. rd: BytesMut::new(),
13. wr: BytesMut::new(),
14. }
15. }
16. }

从套接字读取的数据缓冲到rd中。读取完整行后，将返回给调用者。调用者提交以写入套接字的行被缓冲到wr中，然后刷新。

这是读取一半的实现方式：

1. impl Stream for Lines {
2. type Item = BytesMut;
3. type Error = io::Error;

5. fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error> {

11. let sock_closed = self.fill_read_buf()?.is_ready();

14. let pos = self.rd.windows(2)
15. .position(|bytes| bytes == b"\r\n");

17. if let Some(pos) = pos {

20. let mut line = self.rd.split_to(pos + 2);

23. line.split_off(pos);

26. return Ok(Async::Ready(Some(line)));
27. }

29. if sock_closed {
30. Ok(Async::Ready(None))
31. } else {
32. Ok(Async::NotReady)
33. }
34. }
35. }

37. impl Lines {
38. fn fill_read_buf(&mut self) -> Result<Async<()>, io::Error> {
39. loop {

43. self.rd.reserve(1024);

48. let n = try_ready!(self.socket.read_buf(&mut self.rd));

50. if n == 0 {
51. return Ok(Async::Ready(()));
52. }
53. }
54. }
55. }

该示例使用字节包中的BytesMut。这为在网络环境中处理字节序列提供了一些很好的实用程序。 Stream实现产生的BytesMut值只包含一行。

与往常一样，实现返回Async的函数的关键是永远不会返回Async :: NotReady，除非函数实现收到Async :: NotReady本身。在此示例中，仅当fill_read_buf返回NotReady时才返回NotReady，如果TcpStream :: read_buf返回NotReady，则fill_read_buf仅返回NotReady。

1. struct Lines {
2. socket: TcpStream,
3. rd: BytesMut,
4. wr: BytesMut,
5. }
6. impl Lines {
7. fn buffer(&mut self, line: &[u8]) {

11. self.wr.put(line);
12. }

14. fn poll_flush(&mut self) -> Poll<(), io::Error> {

16. while !self.wr.is_empty() {

18. let n = try_ready!(self.socket.poll_write(&self.wr));

22. assert!(n > 0);

25. let _ = self.wr.split_to(n);
26. }

28. Ok(Async::Ready(()))
29. }
30. }
31. fn main() {}

调用者通过调用缓冲区对所有行进行排队。这会将该行附加到内部wr缓冲区。然后，一旦所有数据排队，调用者就会调用poll_flush，它会对套接字进行实际写入操作。 poll_flush仅在所有排队数据成功写入套接字后才返回Ready。

与读取半部分类似，仅在函数实现收到NotReady本身时返回NotReady。

Lines编解码器在进程函数中使用如下：

1. fn process(socket: TcpStream, state: Arc<Mutex<Shared>>) {

3. let lines = Lines::new(socket);

13. let connection = lines.into_future()

16. .map_err(|(e, _)| e)

18. .and_then(|(name, lines)| {
19. let name = match name {
20. Some(name) => name,
21. None => {

24. unimplemented!();
25. }
26. };

29. });
30. }

广播消息

下一步是实现处理实际聊天功能的连接处理逻辑，即从一个客户端向所有其他客户端广播消息。

为了实现这一点，我们将明确地实现一个Future，它接受Lines编解码器实例并处理广播逻辑。这个逻辑处理：

在其消息通道上接收消息并将其写入套接字。
从套接字接收消息并将其广播给所有 Peer。完全使用组合器实现此逻辑也是可能的，但需要使用拆分，但尚未涉及。此外，这提供了一个机会，可以看到如何手动实现一个非平凡的 future。

以下是处理连接的广播逻辑的 future定义：

1. struct Peer {

3. name: BytesMut,

6. lines: Lines,

9. state: Arc<Mutex<Shared>>,

15. rx: Rx,

22. addr: SocketAddr,
23. }

并且创建如下Peer实例：

1. impl Peer {
2. fn new(name: BytesMut,
3. state: Arc<Mutex<Shared>>,
4. lines: Lines) -> Peer
5. {

7. let addr = lines.socket.peer_addr().unwrap();

10. let (tx, rx) = mpsc::unbounded();

13. state.lock().unwrap()
14. .peers.insert(addr, tx);

16. Peer {
17. name,
18. lines,
19. state,
20. rx,
21. addr,
22. }
23. }
24. }

为其他 Peer创建mpsc通道，以将其消息发送到此新创建的 Peer。在创建信道之后，将发送半部分插入 Peer映射中。此条目在Peer的drop实现中删除。

1. impl Drop for Peer {
2. fn drop(&mut self) {
3. self.state.lock().unwrap().peers
4. .remove(&self.addr);
5. }
6. }

这是实现

1. impl Future for Peer {
2. type Item = ();
3. type Error = io::Error;

5. fn poll(&mut self) -> Poll<(), io::Error> {

7. loop {

10. match self.rx.poll().unwrap() {
11. Async::Ready(Some(v)) => {

15. self.lines.buffer(&v);
16. }
17. _ => break,
18. }
19. }

22. let _ = self.lines.poll_flush()?;

25. while let Async::Ready(line) = self.lines.poll()? {
26. println!("Received line ({:?}) : {:?}", self.name, line);

28. if let Some(message) = line {

30. let mut line = self.name.clone();
31. line.put(": ");
32. line.put(&message);
33. line.put("\r\n");

41. let line = line.freeze();

44. for (addr, tx) in &self.state.lock().unwrap().peers {

46. if *addr != self.addr {

51. tx.unbounded_send(line.clone()).unwrap();
52. }
53. }
54. } else {

57. return Ok(Async::Ready(()));
58. }
59. }

66. Ok(Async::NotReady)
67. }
68. }

剩下的就是连接刚刚实施的Peer future。为此，将客户端连接任务（在process函数中定义）扩展为使用Peer。

1. let connection = lines.into_future()
2. .map_err(|(e, _)| e)
3. .and_then(|(name, lines)| {

19. let name = match name {
20. Some(name) => name,
21. None => {

24. return Either::A(future::ok(()));
25. }
26. };

28. println!("`{:?}` is joining the chat", name);

34. let peer = Peer::new(
35. name,
36. state,
37. lines);

40. Either::B(peer)
41. })

44. .map_err(|e| {
45. println!("connection error = {:?}", e);
46. });

除了添加Peer之外，还会处理name == None。在这种情况下，远程客户端在识别自身之前终止。

返回多个 future（name == None handler和 Peer）通过将返回的 future包装在Either中来处理。要么是枚举，要为每个变体接受不同的 future类型。这允许返回多个 future类型而不到达trait对象。