# Simple Chat Server using Rust::Tokio
###### tags: `concurrency` `Rust` `tokio` `asynchrnous`

> This is tutorial for creating simple server using Tokio, Rust library for asynchronous programming. Rust is a language which can give safety in terms of _memory_. This also applies for _asynchronous_ programming.

What I am trying to do is creating a simple server that can handle multiple clients. Clients can connect with a server and send messages among them.

## Listener for server

First we have to create _listener_ that listens to connections. It has to be bounded to certain _address_ and accept.

### Code

1. `TcpListener::bind()` creates a new TcpListener that is bounded to specific address. Since it is async function, it returns a type that implements Future trait `impl Future<Output=()>` . To be executed, `.await()` is needed. It applies for every async function.

2. The line of `listener::accept().await` accepts the connection from the address that has been specified. It returns Result<()> so Error has to be handled. If it accepts, `(TcpStream to communicate, address of client)` is returned. This can be used later section.

3. Returned socket should be mutable to be readable and writable.

const ADDR: &str = "localhost:8000"; // Your own address

#[tokio::main]
async fn main() -> std::io::Result<(),()> {
    // 1
    let listener = tokio::TcpListener::bind(ADDR).await?;
    // 2
    match litener.accept().await {
        // 3 
        Ok((mut socket, addr)) => {
            // do something
        }
        
        Err(err) => println!("Error connecting. {}", err);
    }
}

## Echo from server

Now let's create a piece of program(server) that can just echo what comes from client that has been connected. To connect with server, I have used `telnet`.

### Code

1. First we have to deconstruct _socket_ into _reader_ and _writer_. If we use single _socket_ type, we might face the most common compile error, **use-of-moved-value**. Trust me 😉

2. Create some buffer that can read from client message. Here, I have used _BufReader_ from tokio::io.

3. Writer is a response from server. `write_all` writes simply what has been sent from client.

Problem for this server is it only serves one client and handle only one message. To fix this, we need infinite `loop` statement.

#[tokio::main] 
async fn main() -> Result<(),()> {

    let listener = tokio::TcpListener::bind(ADDR).await?;
    Ok((mut socket, addr)) => {
        // To see who's connected
        println!("{:?}", addr)
        // 1
        let (reader, mut writer) = socket.split();
        let mut reader = tokio::io::BufReader::new(reader);
        let mut msg: String = String::new();
        match reader.read_line(&mut msg).await {
            Ok(bytes_size) => {
                // bytes_size can be used somewhere..
                match writer.write_all(&msg.as_bytes()).await {
                    Ok(()) => (),
                    Err(()) => { error handling }
                }
            }
            Err(()) => { error handling }
        }
    }
    
    Err(err) => { error handling }
}

## Handling Multiple Clients & Messages

Currently, server serves only one client. After it accepts from client, reads what has been sent and writes back what has sent and completes its task. To receive multiple connection from client, we need certain trick. `loop`

#[tokio::main]
async fn main() -> Result<(), ()> {
    let listener = tokio::TcpListener::bind(ADDR).await?;
    // To listen for multiple clients
    loop {
        match listener.accept().await {
            Ok() => {}
            Err() => {}
        }
    }
    
}

Now our server can serve multiple clients from bounded address. Let's make a server that can serve multiple messages from multiple clients. When we look at the code what we have written so far, our server can serve as much clients as it can, but has little problem. Whoever connects to server first will block other clients who want to connect. To make our server work concurrently, we need little trick. `tokio::spawn()`. It spawns new thread and run concurrently.
> Problem for current server

#[tokio::main]
async fn main() -> Result<(), ()> {
    let listener = tokio::TcpListener::bind(ADDR).await?;
    // To listen for multiple clients
    loop {
        match listener.accept().await {
            // Other clients will be blocked whenever one client is connected. Code below will not be worked for other clients. 
            Ok((mut socket, addr)) => {
                let (reader, mut writer) = socket.split();
                ...
            }
            Err(()) => {}
        }
    }
            Err() => {}
        }
    }
    
}

> Let's spawn new thread to serve each client!

#[tokio::main]
async fn main() -> Result<(),()> {
    let listener = tokio::TcpListener::bind(ADDR).await?;
    loop {
        match listener.accept().await {
            Ok((socket, addr)) => {
                // spawning new thread for each client
                tokio::spawn(async move {
                    // reader, writer for each client
                    let (reader, writer) = socket.split();
                    ...
                })
            },
            Err(())
        }
    }
}

> Handling multiple messages from multiple clients

#[tokio::main]
async fn main() -> Result<(), ()> {
    let listener = tokio::TcpListener::bind(ADDR).await?;
    loop {
        match listener.accept().await {
            Ok((socket, addr)) => {
                // spawning new thread for each client
                tokio::spawn(async move {
                    // reader, writer for each client
                    // Same as above
                    // Now our server can handle multiple messages from multiple clients
                    loop {
                        match reader.read_line(&msg).await {
                            Ok(()) => Ok(()),
                            Err(()) => Err(())
                        }
                    }
                    ...
                })
            },
            Err(())
        }
    }
}

## Communication Channel
Now our server can handle *multiple clients & multiple messages*. However, it is just echoing over and over. We need a server that can communicate among clients! To achieve this, we need another features of ***tokio broadcast channel***. Moreover, to run multiple *Future* concurrently ion the same thread, we need ***tokio::select!***. It makes multiple *future* run concurrenlty on same thread. Therefore, if one task blocks the thread, other future will not be run. Let's look at the code implementation how they are used in our *Simple Chat Server*.

### Code
1. Create channel for communication between clients that connect each other within server
2. Cloning each channel for each client
3. Make 'read_line and sending message' from client 'receiving message' run concurrently

#[tokio::main]
async fn main() {
    
    // Create listener instance that bounds to certain address
    let listener = tokio::TcpListener::bind(ADDR).await?;
    // Create communication-channel between clients which connect each with server
    let (tx, _) = tokio::sync::broadcast::channel(10);
    loop {
        match listener.accept().await {
            Ok((mut socket, addr)) => {
                // Cloning (tx, rx) for each client
                // rx should be mutable 
                let tx = tx.clone();
                let mut rx = tx.subscribe();
                // Handling task for each client on different thread    
                // Ownership of the environment should be moved to the closure. Ex) tx, rx, socket, addr etc..
                tokio::spawn(async move {
                    let (r,w) = socket.split();
                    let mut r = BufReader::new(r);
                    let text = String::new();
                    // Reading line from client & receiving message should run concurrently
                    // On same thread, 'r.read_line' & 'rx.recv()' are running concurrently. After one task is done, remaining task will run 
                    // For example, if 'Client 1' send message 'r.read_line() will call first. 
                    // While 'rx.recv()' will be called for other 'Client' and vise versa.
                    tokio::select! {
                        result = r.read_line(&mut text) => {
                            match result {
                                Ok(_) => {
                                    tx.send((text.clone(), addr )).unwrap();
                                },
                                Err(()) => ()
                            }
                        }
                        result = rx.recv() => {
                            let (msg, recv_addr) = result.unwrap();
                            // For preventing echoing from 'server'
                            if recv_addr != addr {
                                match writer.write_all(msg.as_bytes()).await {
                                    Ok(()) => (),
                                    Err(()) => ()
                                }
                            }
                        }
                    }
                })
            },
            Err(()) => {}
        }
    }
}

# Conclusion
For this article, I have used *tokio* crate to handle asynchrnous task. To connect multiple clients, wrapping asynchronous code into ***loop*** block. To run on multiple thread, we need to spawn a new thread to run with ***tokio::spawn***. Most importantly, to move environment variable ownership, ***move*** syntax is needed. To run multiple futures concurrently, wrap multiple async function with ***tokio::select!***. It was my toy-project to get used to with *tokio* and I found that when writing asynchronous code, it is the powerful tool to make it happen and work very smoothly. Next project will be about creating p2p server with ***libp2p***

Check [Github](https://github.com/cocoyoon/dev-playground/tree/main/concurrency/asynchronous/simple-chat) for full code implementation.

Contact
[LinkedIn](https://www.linkedin.com/in/soyoun-jeong-066165179/) | [Twitter](https://twitter.com/proofofyoon)

# Reference
[Creating a Chat Server with async Rust and Tokio](https://www.youtube.com/watch?v=T2mWg91sx-o)