In this tutorial, we will implement an asynchronous chat on top of async-std.
The chat uses a simple text protocol over TCP.
The protocol consists of utf-8 messages, separated by \n
.
The client connects to the server and sends login as a first line. After that, the client can send messages to other clients using the following syntax:
login1, login2, ... login2: message
Each of the specified clients than receives a from login: message
message.
A possible session might look like this
alice
> alice
> bob: hello
< from bob: hi! |
bob
> bob
< from alice: hello
> alice, bob: hi!
< from bob: hi! |
The main challenge for the chat server is keeping track of many concurrent connections. The main challenge for the chat client is managing concurrent outgoing messages, incoming messages and user’s typing.
Let’s create a new Cargo project:
$ cargo new a-chat
$ cd a-chat
Then add the following to your Cargo.toml
:
futures-preview = { version = "0.3.0-alpha.18", features = [ "async-await", "nightly" ] }
async-std = "0.99"
At the moment async-std
requires nightly, so let’s add a rustup override for convenience:
$ rustup override add nightly
$ rustc --version
rustc 1.38.0-nightly (c4715198b 2019-08-05)
Let’s implement the scaffold of the server: a loop that binds a TCP socket to an address and starts accepting connections.
First of all, let’s add required import boilerplate:
use std::net::ToSocketAddrs; (1)
use async_std::{
prelude::*, (2)
task, (3)
net::TcpListener, (4)
};
type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>; (5)
-
async_std
usesstd
types where appropriate. We’ll needToSocketAddrs
to specify address to listen on. -
prelude
re-exports some traits required to work with futures and streams -
The
task
module roughtly corresponds tostd::thread
module, but tasks are much lighter weight. A single thread can run many tasks. -
For the socket type, we use
TcpListener
fromasync_std
, which is just likestd::net::TcpListener
, but is non-blocking and usesasync
API. -
We will skip implementing comprehensive error handling in this example. To propagate the errors, we will use a boxed error trait object.
NoteDo you know that there’s From<&'_ str> for Box<dyn Error>
implementation in stdlib, which allows you to use strings with?
operator?
Now we can write the server’s accept loop:
async fn server(addr: impl ToSocketAddrs) -> Result<()> { (1)
let listener = TcpListener::bind(addr).await?; (2)
let mut incoming = listener.incoming();
while let Some(stream) = incoming.next().await { (3)
// TODO
}
Ok(())
}
-
We mark
server
function asasync
, which allows us to use.await
syntax inside. -
TcpListener::bind
call returns a future, which we.await
to extract theResult
, and then?
to get aTcpListener
. Note how.await
and?
work nicely together. This is exactly howstd::net::TcpListener
works, but with.await
added. Mirroring API ofstd
is an explicit design goal ofasync_std
. -
Here, we would like to iterate incoming sockets, just how one would do in
std
:let listener: std::net::TcpListener = unimplemented!(); for stream in listener.incoming() { }
Unfortunately this doesn’t quite work with
async
yet, because there’s no support forasync
for-loops in the language yet. For this reason we have to implement the loop manually, by usingwhile let Some(item) = iter.next().await
pattern.
Finally, let’s add main:
fn main() -> Result<()> {
let fut = server("127.0.0.1:8080");
task::block_on(fut)
}
The crucial thing to realise that is in Rust, unlike other languages, calling an async function does not run any code.
Async functions only construct futures, which are inert state machines.
To start stepping through the future state-machine in an async function, you should use .await
.
In a non-async function, a way to execute a future is to handle it to the executor.
In this case, we use task::block_on
to execute future on the current thread and block until it’s done.
Let’s implement the receiving part of the protocol. We need to:
-
split incoming
TcpStream
on\n
and decode bytes as utf-8 -
interpret the first line as a login
-
parse the rest of the lines as a
login: message
use async_std::net::TcpStream;
async fn server(addr: impl ToSocketAddrs) -> Result<()> {
let listener = TcpListener::bind(addr).await?;
let mut incoming = listener.incoming();
while let Some(stream) = incoming.next().await {
let stream = stream?;
println!("Accepting from: {}", stream.peer_addr()?);
let _handle = task::spawn(client(stream)); (1)
}
Ok(())
}
async fn client(stream: TcpStream) -> Result<()> {
let reader = BufReader::new(&stream); (2)
let mut lines = reader.lines();
let name = match lines.next().await { (3)
None => Err("peer disconnected immediately")?,
Some(line) => line?,
};
println!("name = {}", name);
while let Some(line) = lines.next().await { (4)
let line = line?;
let (dest, msg) = match line.find(':') { (5)
None => continue,
Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
};
let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
let msg: String = msg.trim().to_string();
}
Ok(())
}
-
We use
task::spawn
function to spawn an independent task for working with each client. That is, after accepting the client theserver
loop immediately starts waiting for the next one. This is the core benefit of event-driven architecture: we serve many number of clients concurrently, without spending many hardware threads. -
Luckily, the "split byte stream into lines" functionality is already implemented.
.lines()
call returns a stream ofString
's. TODO: show how one would implementlines
by hand? -
We get the first line — login
-
And, once again, we implement a manual async for loop.
-
Finally, we parse each line into a list of destination logins and the message itself.
One serious problem in the above solution is that, while we correctly propagate errors in the client
, we just drop the error on the floor afterwards!
That is, task::spawn
does not return error immediately (it can’t, it needs to run the future to completion first), only after it is joined.
We can "fix" it by waiting for the task to be joined, like this:
let handle = task::spawn(client(stream)); (1)
handle.await?
The .await
waits until the client finishes, and ?
propagates the result.
There are two problems with this solution however! First, because we immediately await the client, we can only handle one client at time, and that completely defeats the purpose of async! Second, if a client encounters an IO error, the whole server immediately exits. That is, a flaky internet connection of one peer brings down the whole chat room!
A correct way to handle client errors in this case is log them, and continue serving other clients. So let’s use a helper function for this:
fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
where
F: Future<Output = Result<()>> + Send + 'static,
{
task::spawn(async move {
if let Err(e) = fut.await {
eprintln!("{}", e)
}
})
}
Now it’s time to implement the other half — sending messages.
A most obvious way to implement sending is to give each client
access to the write half of TcpStream
of each other clients.
That way, a client can directly .write_all
a message to recipients.
However, this would be wrong: if Alice sends bob: foo
, and Charley sends bob: bar
, Bob might actually receive fobaor
.
Sending a message over a socket might require several syscalls, so two concurrent .write_all
's might interfere with each other!
As a rule of thumb, only a single task should write to each TcpStream
.
So let’s create a client_writer
task which receives messages over a channel and writes them to the socket.
This task would be the point of serialization of messages.
if Alice and Charley send two messages to Bob at the same time, Bob will see the messages in the same order as they arrive in the channel.
use futures::channel::mpsc; (1)
use futures::SinkExt;
type Sender<T> = mpsc::UnboundedSender<T>; (2)
type Receiver<T> = mpsc::UnboundedReceiver<T>;
async fn client_writer(
mut messages: Receiver<String>,
stream: Arc<TcpStream>, (3)
) -> Result<()> {
let mut stream = &*stream;
while let Some(msg) = messages.next().await {
stream.write_all(msg.as_bytes()).await?;
}
Ok(())
}
-
We will use channels from the
futures
crate. -
For simplicity, we will use
unbounded
channels, and won’t be discussing backpressure in this tutorial. -
As
client
andclient_writer
share the sameTcpStream
, we need to put it into anArc
. Note that becauseclient
only reads from andclient_writer
only writes to the stream, so we don’t get a race here.
So how we make sure that messages read in client
flow into the relevant client_writer
?
We should somehow maintain an peers: HashMap<String, Sender<String>>
map which allows a client to find destination channels.
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.
One trick to make reasoning about state simpler comes from the actor model.
We can create a dedicated broker tasks which owns the peers
map and communicates with other tasks by channels.
By hiding peers
inside such "actor" task, we remove the need for mutxes and also make serialization point explicit.
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.
#[derive(Debug)]
enum Event { (1)
NewPeer {
name: String,
stream: Arc<TcpStream>,
},
Message {
from: String,
to: Vec<String>,
msg: String,
},
}
async fn broker(mut events: Receiver<Event>) -> Result<()> {
let mut peers: HashMap<String, Sender<String>> = HashMap::new(); (2)
while let Some(event) = events.next().await {
match event {
Event::Message { from, to, msg } => { (3)
for addr in to {
if let Some(peer) = peers.get_mut(&addr) {
peer.send(format!("from {}: {}\n", from, msg)).await?
}
}
}
Event::NewPeer { name, stream } => {
match peers.entry(name) {
Entry::Occupied(..) => (),
Entry::Vacant(entry) => {
let (client_sender, client_receiver) = mpsc::unbounded();
entry.insert(client_sender); (4)
spawn_and_log_error(client_writer(client_receiver, stream)); (5)
}
}
}
}
}
Ok(())
}
-
Broker should handle two types of events: a message or an arrival of a new peer.
-
Internal state of the broker is a
HashMap
. Note how we don’t need aMutex
here and can confidently say, at each iteration of the broker’s loop, what is the current set of peers -
To handle a message we send it over a channel to each destination
-
To handle new peer, we first register it in the peer’s map …
-
… and then spawn a dedicated task to actually write the messages to the socket.
At this point, we only need to start broker to get a fully-functioning (in the happy case!) chat:
use std::{
net::ToSocketAddrs,
sync::Arc,
collections::hash_map::{HashMap, Entry},
};
use futures::{
channel::mpsc,
SinkExt,
};
use async_std::{
io::BufReader,
prelude::*,
task,
net::{TcpListener, TcpStream},
};
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>;
fn main() -> Result<()> {
task::block_on(server("127.0.0.1:8080"))
}
async fn server(addr: impl ToSocketAddrs) -> Result<()> {
let listener = TcpListener::bind(addr).await?;
let (broker_sender, broker_receiver) = mpsc::unbounded(); (1)
let _broker_handle = task::spawn(broker(broker_receiver));
let mut incoming = listener.incoming();
while let Some(stream) = incoming.next().await {
let stream = stream?;
println!("Accepting from: {}", stream.peer_addr()?);
spawn_and_log_error(client(broker_sender.clone(), stream));
}
Ok(())
}
async fn client(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
let stream = Arc::new(stream); (2)
let reader = BufReader::new(&*stream);
let mut lines = reader.lines();
let name = match lines.next().await {
None => Err("peer disconnected immediately")?,
Some(line) => line?,
};
broker.send(Event::NewPeer { name: name.clone(), stream: Arc::clone(&stream) }).await (3)
.unwrap();
while let Some(line) = lines.next().await {
let line = line?;
let (dest, msg) = match line.find(':') {
None => continue,
Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
};
let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
let msg: String = msg.trim().to_string();
broker.send(Event::Message { (4)
from: name.clone(),
to: dest,
msg,
}).await.unwrap();
}
Ok(())
}
async fn client_writer(
mut messages: Receiver<String>,
stream: Arc<TcpStream>,
) -> Result<()> {
let mut stream = &*stream;
while let Some(msg) = messages.next().await {
stream.write_all(msg.as_bytes()).await?;
}
Ok(())
}
#[derive(Debug)]
enum Event {
NewPeer {
name: String,
stream: Arc<TcpStream>,
},
Message {
from: String,
to: Vec<String>,
msg: String,
},
}
async fn broker(mut events: Receiver<Event>) -> Result<()> {
let mut peers: HashMap<String, Sender<String>> = HashMap::new();
while let Some(event) = events.next().await {
match event {
Event::Message { from, to, msg } => {
for addr in to {
if let Some(peer) = peers.get_mut(&addr) {
peer.send(format!("from {}: {}\n", from, msg)).await?
}
}
}
Event::NewPeer { name, stream} => {
match peers.entry(name) {
Entry::Occupied(..) => (),
Entry::Vacant(entry) => {
let (client_sender, client_receiver) = mpsc::unbounded();
entry.insert(client_sender); (4)
spawn_and_log_error(client_writer(client_receiver, stream)); (5)
}
}
}
}
}
Ok(())
}
-
Inside the
server
, we create broker’s channel andtask
. -
Inside
client
, we need to wrapTcpStream
into anArc
, to be able to share it with theclient_writer
. -
On login, we notify the broker. Note that we
.unwrap
on send: broker should outlive all the clients and if that’s not the case the broker probably panicked, so we can escalate the panic as well. -
Similarly, we forward parsed messages to the broker, assuming that it is alive.
On of the problems of the current implementation is that it doesn’t handle graceful shutdown. If we break from the accept loop for some reason, all in-flight tasks are just dropped on the floor. A more correct shutdown sequence would be:
-
Stop accepting new clients
-
Deliver all pending messages
-
Exit the process
A clean shutdown in a channel based architecture is easy, although it can appear a magic trick at first.
In Rust, receiver side of a channel is closed as soon as all senders are dropped.
That is, as soon as producers exit and drop their senders, the rest of the system shutdowns naturally.
In async_std
this translates to two rules:
-
Make sure that channels form an acyclic graph.
-
Take care to wait, in the correct order, until intermediate layers of the system process pending messages.
In a-chat
, we already have an unidirectional flow of messages: reader → broker → writer
.
However, we never wait for broker and writers, which might cause some messages to get dropped.
Let’s add waiting to the server:
async fn server(addr: impl ToSocketAddrs) -> Result<()> {
let listener = TcpListener::bind(addr).await?;
let (broker_sender, broker_receiver) = mpsc::unbounded();
let broker = task::spawn(broker(broker_receiver));
let mut incoming = listener.incoming();
while let Some(stream) = incoming.next().await {
let stream = stream?;
println!("Accepting from: {}", stream.peer_addr()?);
spawn_and_log_error(client(broker_sender.clone(), stream));
}
drop(broker_sender); (1)
broker.await?; (5)
Ok(())
}
And to the broker:
async fn broker(mut events: Receiver<Event>) -> Result<()> {
let mut writers = Vec::new();
let mut peers: HashMap<String, Sender<String>> = HashMap::new();
while let Some(event) = events.next().await { (2)
match event {
Event::Message { from, to, msg } => {
for addr in to {
if let Some(peer) = peers.get_mut(&addr) {
peer.send(format!("from {}: {}\n", from, msg)).await?
}
}
}
Event::NewPeer { name, stream} => {
match peers.entry(name) {
Entry::Occupied(..) => (),
Entry::Vacant(entry) => {
let (client_sender, client_receiver) = mpsc::unbounded();
entry.insert(client_sender);
let handle = spawn_and_log_error(client_writer(client_receiver, stream));
writers.push(handle); (4)
}
}
}
}
}
drop(peers); (3)
for writer in writers { (4)
writer.await?;
}
Ok(())
}
Notice what happens with all of the channels once we exit the accept loop:
-
First, we drop the main broker’s sender. That way when the readers are done, there’s no sender for the broker’s channel, and the chanel closes.
-
Next, the broker exits
while let Some(event) = events.next().await
loop. -
It’s crucial that, at this stage, we drop the
peers
map. This drops writer’s senders. -
Now we can join all of the writers.
-
Finally, we join the broker, which also guarantees that all the writes have terminated.
Currently, we only ever add new peers to the map. This is clearly wrong: if a peer closes connection to the chat, we should not try to send any more messages to it.
One subtlety with handling disconnection is that we can detect it either in the reader’s task, or in the writer’s task.
The most obvious solution here is to just remove the peer from the peers
map in both cases, but this would be wrong.
If both read and write fail, we’ll remove the peer twice, but it can be the case that the peer reconnected between the two failures!
To fix this, we will only remove the peer when the write side finishes.
If the read side finishes we will notify the write side that it should stop as well.
That is, we need to add an ability to signal shutdown for the writer task.
One way to approach this is a shutdown: Receiver<()>
channel.
There’s a more minimal solution however, which makes a clever use of RAII.
Closing a channel is a synchronization event, so we don’t need to send a shutdown message, we can just drop the sender.
This way, we statically guarantee that we issue shutdown exactly once, even if we early return via ?
or panic.
First, let’s add shutdown channel to the client
:
#[derive(Debug)]
enum Void {} (1)
#[derive(Debug)]
enum Event {
NewPeer {
name: String,
stream: Arc<TcpStream>,
shutdown: Receiver<Void>, (2)
},
Message {
from: String,
to: Vec<String>,
msg: String,
},
}
async fn client(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
// ...
let (_shutdown_sender, shutdown_receiver) = mpsc::unbounded::<Void>(); (3)
broker.send(Event::NewPeer {
name: name.clone(),
stream: Arc::clone(&stream),
shutdown: shutdown_receiver,
}).await.unwrap();
// ...
}
-
To enforce that no messages are send along the shutdown channel, we use an uninhabited type.
-
We pass the shutdown channel to the writer task
-
In the reader, we create an
_shutdown_sender
whose only purpose is to get dropped.
In the client_writer
, we now need to chose between shutdown and message channels.
We use select
macro for this purpose:
use futures::select;
async fn client_writer(
messages: &mut Receiver<String>,
stream: Arc<TcpStream>,
mut shutdown: Receiver<Void>, (1)
) -> Result<()> {
let mut stream = &*stream;
loop { (2)
select! {
msg = messages.next() => match msg {
Some(msg) => stream.write_all(msg.as_bytes()).await?,
None => break,
},
void = shutdown.next() => match void {
Some(void) => match void {}, (3)
None => break,
}
}
}
Ok(())
}
-
We add shutdown channel as an argument.
-
Because of
select
, we can’t use awhile let
loop, so we desugar it further into aloop
. -
In the shutdown case we use
match void {}
as a statically-checkedunreachable!()
.
Another problem is that between the moment we detect disconnection in client_writer
and the moment when we actually remove the peer from the peers
map, new messages might be pushed into the peer’s channel.
To not lose these messages completely, we’ll return the messages channel back to broker.
This also allows us to establish a useful invariant that the message channel strictly outlives the peer in the peers
map, and make the broker itself infailable.
The final code looks like this:
use std::{
net::ToSocketAddrs,
sync::Arc,
collections::hash_map::{HashMap, Entry},
};
use futures::{
channel::mpsc,
SinkExt,
select,
};
use async_std::{
io::BufReader,
prelude::*,
task,
net::{TcpListener, TcpStream},
};
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 {}
fn main() -> Result<()> {
task::block_on(server("127.0.0.1:8080"))
}
async fn server(addr: impl ToSocketAddrs) -> Result<()> {
let listener = TcpListener::bind(addr).await?;
let (broker_sender, broker_receiver) = mpsc::unbounded();
let broker = task::spawn(broker(broker_receiver));
let mut incoming = listener.incoming();
while let Some(stream) = incoming.next().await {
let stream = stream?;
println!("Accepting from: {}", stream.peer_addr()?);
spawn_and_log_error(client(broker_sender.clone(), stream));
}
drop(broker_sender);
broker.await;
Ok(())
}
async fn client(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
let stream = Arc::new(stream);
let reader = BufReader::new(&*stream);
let mut lines = reader.lines();
let name = match lines.next().await {
None => Err("peer disconnected immediately")?,
Some(line) => line?,
};
let (_shutdown_sender, shutdown_receiver) = mpsc::unbounded::<Void>();
broker.send(Event::NewPeer {
name: name.clone(),
stream: Arc::clone(&stream),
shutdown: shutdown_receiver,
}).await.unwrap();
while let Some(line) = lines.next().await {
let line = line?;
let (dest, msg) = match line.find(':') {
None => continue,
Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
};
let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
let msg: String = msg.trim().to_string();
broker.send(Event::Message {
from: name.clone(),
to: dest,
msg,
}).await.unwrap();
}
Ok(())
}
async fn client_writer(
messages: &mut Receiver<String>,
stream: Arc<TcpStream>,
mut shutdown: Receiver<Void>,
) -> Result<()> {
let mut stream = &*stream;
loop {
select! {
msg = messages.next() => match msg {
Some(msg) => stream.write_all(msg.as_bytes()).await?,
None => break,
},
void = shutdown.next() => match void {
Some(void) => match void {},
None => break,
}
}
}
Ok(())
}
#[derive(Debug)]
enum Event {
NewPeer {
name: String,
stream: Arc<TcpStream>,
shutdown: Receiver<Void>,
},
Message {
from: String,
to: Vec<String>,
msg: String,
},
}
async fn broker(mut events: Receiver<Event>) {
let (disconnect_sender, mut disconnect_receiver) = (1)
mpsc::unbounded::<(String, Receiver<String>)>();
let mut peers: HashMap<String, Sender<String>> = HashMap::new();
loop {
let event = select! {
event = events.next() => match event {
None => break, (2)
Some(event) => event,
},
disconnect = disconnect_receiver.next() => {
let (name, _pending_messages) = disconnect.unwrap(); (3)
assert!(peers.remove(&name).is_some());
continue;
},
};
match event {
Event::Message { from, to, msg } => {
for addr in to {
if let Some(peer) = peers.get_mut(&addr) {
peer.send(format!("from {}: {}\n", from, msg)).await
.unwrap() (6)
}
}
}
Event::NewPeer { name, stream, shutdown } => {
match peers.entry(name.clone()) {
Entry::Occupied(..) => (),
Entry::Vacant(entry) => {
let (client_sender, mut client_receiver) = mpsc::unbounded();
entry.insert(client_sender);
let mut disconnect_sender = disconnect_sender.clone();
spawn_and_log_error(async move {
let res = client_writer(&mut client_receiver, stream, shutdown).await;
disconnect_sender.send((name, client_receiver)).await (4)
.unwrap();
res
});
}
}
}
}
}
drop(peers); (5)
drop(disconnect_sender); (6)
while let Some((_name, _pending_messages)) = disconnect_receiver.next().await {
}
}
fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
where
F: Future<Output = Result<()>> + Send + 'static,
{
task::spawn(async move {
if let Err(e) = fut.await {
eprintln!("{}", e)
}
})
}
-
In the broker, we create a channel to reap disconnected peers and their undelivered messages.
-
The broker’s main loop exits when the input events channel is exhausted (that is, when all readers exit).
-
Because broker itself holds a
disconnect_sender
, we know that the disconnections channel can’t be fully drained in the main loop. -
We send peer’s name and pending messages to the disconnections channel in both the happy and the not-so-happy path. Again, we can safely unwrap because broker outlives writers.
-
We drop
peers
map to close writers' messages channel and shut down the writers for sure. It is not strictly necessary in the current setup, where the broker waits for readers' shutdown anyway. However, if we add a server-initiated shutdown (for example, kbd:[ctrl+c] handling), this will be a way for the broker to shutdown the writers. -
Finally, we close and drain the disconnections channel.
Let’s now implement the client for the chat. Because the protocol is line-based, the implementation is pretty straightforward:
-
Lines read from stdin should be send over the socket.
-
Lines read from the socket should be echoed to stdout.
Unlike the server, the client needs only limited concurrency, as it interacts with only a single user. For this reason, async doesn’t bring a lot of performance benefits in this case.
However, async is still useful for managing concurrency!
Specifically, the client should simultaneously read from stdin and from the socket.
Programming this with threads is cumbersome, especially when implementing clean shutdown.
With async, we can just use the select!
macro.
use std::net::ToSocketAddrs;
use futures::select;
use async_std::{
prelude::*,
net::TcpStream,
task,
io::{stdin, BufReader},
};
type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
fn main() -> Result<()> {
task::block_on(try_main("127.0.0.1:8080"))
}
async fn try_main(addr: impl ToSocketAddrs) -> Result<()> {
let stream = TcpStream::connect(addr).await?;
let (reader, mut writer) = (&stream, &stream); (1)
let reader = BufReader::new(reader);
let mut lines_from_server = futures::StreamExt::fuse(reader.lines()); (2)
let stdin = BufReader::new(stdin());
let mut lines_from_stdin = futures::StreamExt::fuse(stdin.lines()); (2)
loop {
select! { (3)
line = lines_from_server.next() => match line {
Some(line) => {
let line = line?;
println!("{}", line);
},
None => break,
},
line = lines_from_stdin.next() => match line {
Some(line) => {
let line = line?;
writer.write_all(line.as_bytes()).await?;
writer.write_all(b"\n").await?;
}
None => break,
}
}
}
Ok(())
}
-
Here we split
TcpStream
into read and write halfs: there’simpl AsyncRead for &'_ TcpStream
, just like the one in std. -
We crate a stream of lines for both the socket and stdin.
-
In the main select loop, we print the lines we receive from server and send the lines we read from the console.