Skip to content

Commit ba1ee2d

Browse files
unleashedStjepan Glavina
authored and
Stjepan Glavina
committed
Fix a-chat tutorial issues (#573)
* tutorial/receiving_messages: fix future output type bound * tutorial/receiving_messages: remove unneeded message trimming Trimming was done twice on messages, so one of the two instances can be removed. I personally think removing the first instance, in which we are splitting names from messages makes the code more readable than removing the second instance, but other examples further in the tutorial show the second instance removed. * tutorial/receiving_messages: declare use of TcpStream and io::BufReader Readers couldn't see the `use` lines corresponding to these two structures. * tutorial/connecting_readers_and_writers: typos and grammar fixes * tutorial/all_together: remove unneeded use async_std::io * tutorial: use SinkExt consistently from futures::sink::SinkExt * tutorial/handling_disconnection: hide mpsc use clause and remove empty lines The empty lines translate to the output making it look weird. * tutorial/handling_disconnection: fix typos * tutorial/handling_disconnection: use ? in broker_handle.await We were happy to return an Err variant from the broker_handle before and nothing has changed in this regard, so bubbling it up to run().
1 parent b3d30de commit ba1ee2d

File tree

5 files changed

+26
-23
lines changed

5 files changed

+26
-23
lines changed

docs/src/tutorial/all_together.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,13 @@ At this point, we only need to start the broker to get a fully-functioning (in t
66
# extern crate async_std;
77
# extern crate futures;
88
use async_std::{
9-
io::{self, BufReader},
9+
io::BufReader,
1010
net::{TcpListener, TcpStream, ToSocketAddrs},
1111
prelude::*,
1212
task,
1313
};
1414
use futures::channel::mpsc;
15-
use futures::SinkExt;
15+
use futures::sink::SinkExt;
1616
use std::{
1717
collections::hash_map::{HashMap, Entry},
1818
sync::Arc,

docs/src/tutorial/clean_shutdown.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ Let's add waiting to the server:
3030
# task,
3131
# };
3232
# use futures::channel::mpsc;
33-
# use futures::SinkExt;
33+
# use futures::sink::SinkExt;
3434
# use std::{
3535
# collections::hash_map::{HashMap, Entry},
3636
# sync::Arc,
@@ -163,7 +163,7 @@ And to the broker:
163163
# task,
164164
# };
165165
# use futures::channel::mpsc;
166-
# use futures::SinkExt;
166+
# use futures::sink::SinkExt;
167167
# use std::{
168168
# collections::hash_map::{HashMap, Entry},
169169
# sync::Arc,

docs/src/tutorial/connecting_readers_and_writers.md

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,12 @@
22
## Connecting Readers and Writers
33

44
So how do we make sure that messages read in `connection_loop` flow into the relevant `connection_writer_loop`?
5-
We should somehow maintain an `peers: HashMap<String, Sender<String>>` map which allows a client to find destination channels.
5+
We should somehow maintain a `peers: HashMap<String, Sender<String>>` map which allows a client to find destination channels.
66
However, this map would be a bit of shared mutable state, so we'll have to wrap an `RwLock` over it and answer tough questions of what should happen if the client joins at the same moment as it receives a message.
77

88
One trick to make reasoning about state simpler comes from the actor model.
9-
We can create a dedicated broker tasks which owns the `peers` map and communicates with other tasks by channels.
10-
By hiding `peers` inside such an "actor" task, we remove the need for mutxes and also make serialization point explicit.
9+
We can create a dedicated broker task which owns the `peers` map and communicates with other tasks using channels.
10+
By hiding `peers` inside such an "actor" task, we remove the need for mutexes and also make the serialization point explicit.
1111
The order of events "Bob sends message to Alice" and "Alice joins" is determined by the order of the corresponding events in the broker's event queue.
1212

1313
```rust,edition2018
@@ -92,9 +92,9 @@ async fn broker_loop(mut events: Receiver<Event>) -> Result<()> {
9292
}
9393
```
9494

95-
1. Broker should handle two types of events: a message or an arrival of a new peer.
96-
2. Internal state of the broker is a `HashMap`.
95+
1. The broker task should handle two types of events: a message or an arrival of a new peer.
96+
2. The internal state of the broker is a `HashMap`.
9797
Note how we don't need a `Mutex` here and can confidently say, at each iteration of the broker's loop, what is the current set of peers
9898
3. To handle a message, we send it over a channel to each destination
99-
4. To handle new peer, we first register it in the peer's map ...
99+
4. To handle a new peer, we first register it in the peer's map ...
100100
5. ... and then spawn a dedicated task to actually write the messages to the socket.

docs/src/tutorial/handling_disconnection.md

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ First, let's add a shutdown channel to the `connection_loop`:
2222
# extern crate futures;
2323
# use async_std::net::TcpStream;
2424
# use futures::channel::mpsc;
25-
# use futures::SinkExt;
25+
# use futures::sink::SinkExt;
2626
# use std::sync::Arc;
2727
#
2828
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
@@ -60,8 +60,8 @@ async fn connection_loop(mut broker: Sender<Event>, stream: Arc<TcpStream>) -> R
6060
}
6161
```
6262

63-
1. To enforce that no messages are send along the shutdown channel, we use an uninhabited type.
64-
2. We pass the shutdown channel to the writer task
63+
1. To enforce that no messages are sent along the shutdown channel, we use an uninhabited type.
64+
2. We pass the shutdown channel to the writer task.
6565
3. In the reader, we create a `_shutdown_sender` whose only purpose is to get dropped.
6666

6767
In the `connection_writer_loop`, we now need to choose between shutdown and message channels.
@@ -71,14 +71,12 @@ We use the `select` macro for this purpose:
7171
# extern crate async_std;
7272
# extern crate futures;
7373
# use async_std::{net::TcpStream, prelude::*};
74-
use futures::channel::mpsc;
74+
# use futures::channel::mpsc;
7575
use futures::{select, FutureExt};
7676
# use std::sync::Arc;
77-
7877
# type Receiver<T> = mpsc::UnboundedReceiver<T>;
7978
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
8079
# type Sender<T> = mpsc::UnboundedSender<T>;
81-
8280
# #[derive(Debug)]
8381
# enum Void {} // 1
8482
@@ -112,7 +110,7 @@ async fn connection_writer_loop(
112110

113111
Another problem is that between the moment we detect disconnection in `connection_writer_loop` and the moment when we actually remove the peer from the `peers` map, new messages might be pushed into the peer's channel.
114112
To not lose these messages completely, we'll return the messages channel back to the broker.
115-
This also allows us to establish a useful invariant that the message channel strictly outlives the peer in the `peers` map, and makes the broker itself infailable.
113+
This also allows us to establish a useful invariant that the message channel strictly outlives the peer in the `peers` map, and makes the broker itself infallible.
116114

117115
## Final Code
118116

@@ -128,7 +126,8 @@ use async_std::{
128126
task,
129127
};
130128
use futures::channel::mpsc;
131-
use futures::{select, FutureExt, SinkExt};
129+
use futures::sink::SinkExt;
130+
use futures::{select, FutureExt};
132131
use std::{
133132
collections::hash_map::{Entry, HashMap},
134133
future::Future,
@@ -158,7 +157,7 @@ async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
158157
spawn_and_log_error(connection_loop(broker_sender.clone(), stream));
159158
}
160159
drop(broker_sender);
161-
broker_handle.await;
160+
broker_handle.await?;
162161
Ok(())
163162
}
164163

docs/src/tutorial/receiving_messages.md

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,14 +10,18 @@ We need to:
1010
```rust,edition2018
1111
# extern crate async_std;
1212
# use async_std::{
13-
# io::BufReader,
14-
# net::{TcpListener, TcpStream, ToSocketAddrs},
13+
# net::{TcpListener, ToSocketAddrs},
1514
# prelude::*,
1615
# task,
1716
# };
1817
#
1918
# type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
2019
#
20+
use async_std::{
21+
io::BufReader,
22+
net::TcpStream,
23+
};
24+
2125
async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
2226
let listener = TcpListener::bind(addr).await?;
2327
let mut incoming = listener.incoming();
@@ -46,7 +50,7 @@ async fn connection_loop(stream: TcpStream) -> Result<()> {
4650
Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
4751
};
4852
let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
49-
let msg: String = msg.trim().to_string();
53+
let msg: String = msg.to_string();
5054
}
5155
Ok(())
5256
}
@@ -130,7 +134,7 @@ So let's use a helper function for this:
130134
# };
131135
fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
132136
where
133-
F: Future<Output = io::Result<()>> + Send + 'static,
137+
F: Future<Output = Result<()>> + Send + 'static,
134138
{
135139
task::spawn(async move {
136140
if let Err(e) = fut.await {

0 commit comments

Comments
 (0)