Jan 14, 2024

Asynchronous programming in RUST

Asynchronous programming in Rust is built around the async and await keywords, which allow you to write non-blocking code.

When you mark a function as async, it returns a Future instead of blocking.

You can .await this future to get the result when it is ready.

Tokio

Tokio is a runtime for writing reliable, asynchronous, and scalable applications in Rust.

It provides:

Asynchronous I/O (e.g., TCP, UDP, file I/O)
Timers
Task spawning

Example: Asynchronous TCP Server and Client We'll create a basic example of an asynchronous TCP server and a client using tokio.

Steps

Setup your Rust environment
Add tokio, serde, serde_json to your Cargo.toml
Define commands
Build Server
Test Server

I. Setup your RUST environment

cargo new hash_example
cd hash_example

II. Add crates: sha3 as a dependency

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

III. commands.rs

use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub enum Command {
    GetHelloWorld,
    PostGreeting { message: String },
}

IV. tcp_server.rs

Asynchronous server:

The server uses tokio::net::TcpListener to accept incoming connections.

Each connection is handled in a separate asynchronous task using tokio::spawn.

use crate::commands::Command;
use std::io;
use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
use tokio::net::{TcpListener, TcpStream};

pub async fn start_server(addr: &str) -> io::Result<()> {
    let addr = addr.parse::<SocketAddr>().unwrap();
    let listener = TcpListener::bind(addr).await?;

    loop {
        let (socket, _) = listener.accept().await?;
        tokio::spawn(async move {
            handle_client(socket).await.unwrap();
        });
    }
}

async fn handle_client(socket: TcpStream) -> io::Result<()> {
    let (reader, writer) = socket.into_split();
    let mut reader = BufReader::new(reader);
    let mut writer = BufWriter::new(writer);

    loop {
        let mut buffer = String::new();
        reader.read_line(&mut buffer).await?;

        if buffer.is_empty() {
            break;
        }

        let command: Command = match serde_json::from_str(&buffer) {
            Ok(cmd) => cmd,
            Err(_) => break,
        };

        let response = match command {
            Command::GetHelloWorld => "Hello, world!".to_string(),
            Command::PostGreeting { message } => format!("Received greeting: {}", message),
        };

        writer.write_all(response.as_bytes()).await?;
        writer.flush().await?;
    }

    Ok(())
}

V. main.rs

mod commands;
mod tcp_server;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    tcp_server::start_server("127.0.0.1:8080").await
}

Run issuing below command

cargo run

VI. unit tests bin/client.rs

Asynchronous Client

The client connects to the server, sends commands, and reads responses.

use serde_json::json;
use std::io::{self, BufRead, BufReader, BufWriter, Write};
use std::net::TcpStream;

fn main() -> io::Result<()> {
    let stream = TcpStream::connect("127.0.0.1:8080")?;
    let mut writer = BufWriter::new(&stream);
    let mut reader = BufReader::new(&stream);

    // Send GetHelloWorld command
    let get_hello = json!({"GetHelloWorld": {}}).to_string() + "\n";
    writer.write_all(get_hello.as_bytes())?;
    writer.flush()?;

    // Read the response
    let mut response = String::new();
    reader.read_line(&mut response)?;
    println!("Response: {}", response.trim());

    // Send PostGreeting command
    let post_greeting = json!({"PostGreeting": {"message": "Hello, Rust!"}}).to_string() + "\n";
    writer.write_all(post_greeting.as_bytes())?;
    writer.flush()?;

    // Read the response
    response.clear();
    reader.read_line(&mut response)?;
    println!("Response: {}", response.trim());

    Ok(())
}