Async Rust กับ Tokio

use tokio::runtime;

fn main() {
    // สร้าง runtime แบบ multi-threaded ที่มี 4 worker threads
    let runtime = runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all() // เปิดฟีเจอร์ทั้งหมดที่จำเป็นสำหรับ net, time, ฯลฯ
        .build()
        .expect("Failed to build runtime");

    // รัน future บน runtime ที่เราสร้างขึ้น
    runtime.block_on(async {
        println!("Hello from custom Tokio runtime!");
        let data = fetch_data().await;
        println!("{}", data);
    });
}

#[tokio::main]
async fn main() {
    // สร้าง task ใหม่ที่จะทำงานพร้อมกับ main task
    let handle = tokio::spawn(async {
        println!("Running in a spawned task!");
        // ทำบางอย่างที่ใช้เวลา เช่น รอเวลา
        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
        println!("Task completed!
    });

    // รอให้ task ที่เราสร้างเสร็จสิ้น
    // handle เป็นประเภท JoinHandle ที่ช่วยให้เราสามารถรอผลลัพธ์ได้
    let result = handle.await;
    println!("Task finished with result: {:?}", result);
}

#[tokio::main]
async fn main() {
    let data = vec![1, 2, 3, 4, 5];
    let handle = tokio::spawn(move || async {
        // `data` ถูกย้ายเข้าไปใน task นี้ด้วย move
        let sum: i32 = data.iter().sum();
        println!("Sum: {}", sum);
        sum
    });

    let result = handle.await.unwrap();
    println!("Result from task: {}", result);
}

use tokio::sync::mpsc;
use tokio::time;

#[tokio::main]
async fn main() {
    // สร้าง channel: tx สำหรับส่ง, rx สำหรับรับ
    let (tx, mut rx) = mpsc::channel(32); // 32 คือขนาด buffer

    // สร้างหลาย task ที่ส่งข้อความไปยังตัวรับเดียวกัน
    for i in 0..3 {
        let tx = tx.clone(); // คลอน tx เพื่อให้แต่ละ task มีสำเนาของตัวส่ง
        tokio::spawn(async move {
            // แต่ละ task ส่งข้อความหลายครั้ง
            for j in 0..3 {
                let msg = format!("task {} message {}", i, j);
                if tx.send(msg).await.is_err() {
                    // ตัวรับได้ยกเลิกการทำงานแล้ว
                    eprintln!("Receiver dropped, stopping");
                    break;
                }
                // หน่วงเวลาเล็กน้อยเพื่อให้เห็นการสลับกันทำงาน
                time::sleep(time::Duration::from_millis(100)).await;
            }
        });
    }

    // ตัวรับทำงานใน main task
    // เราจะรับข้อความทั้งหมดแล้วพิมพ์ออกมา
    while let Some(msg) = rx.recv().await {
        println!("Received: {}", msg);
    }
    println!("All messages received.");
}

use tokio::sync::oneshot;
use tokio::time;

#[tokio::main]
async fn main() {
    // สร้าง oneshot channel
    let (tx, rx) = oneshot::channel::<i32>();

    // ส่งค่าจาก task หนึ่ง
    tokio::spawn(async move {
        // ทำบางอย่างที่ใช้เวลา
        time::sleep(time::Duration::from_secs(1)).await;
        // ส่งผลลัพธ์
        let _ = tx.send(42);
    });

    // รอรับผลลัพธ์ใน main task
    match rx.await {
        Ok(value) => println!("Received: {}", value),
        Err(_) => eprintln!("Sender dropped without sending a value"),
    }
}

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> io::Result<()> {
    // ผูกที่อยู่และพอร์ต
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Server listening on 127.0.0.1:8080");

    loop {
        // รอการเชื่อมต่อใหม่เข้ามา (นี้คือ async operation)
        let (socket, addr) = listener.accept().await?;
        println!("Accepted connection from: {}", addr);

        // สำหรับแต่ละการเชื่อมต่อ เราสร้าง task ใหม่เพื่อจัดการมัน
        tokio::spawn(async move {
            if let Err(e) = process_socket(socket).await {
                eprintln!("Error processing socket: {}", e);
            }
        });
    }
}

async fn process_socket(mut socket: TcpStream) -> io::Result<()> {
    let mut buffer = [0; 1024];

    loop {
        // อ่านข้อมูลจาก socket (async)
        let n = match socket.read(&mut buffer).await {
            // ปิดการเชื่อมต่อเมื่ออ่านไม่ได้ข้อมูล (client ปิด connection)
            Ok(0) => return Ok(()),
            Ok(n) => n,
            Err(e) => return Err(e),
        };

        // ส่งข้อมูลที่อ่านได้กลับไปยัง client (echo)
        socket.write_all(&buffer[..n]).await?;
    }
}

use tokio::time::{self, Duration};

#[tokio::main]
async fn main() {
    // สร้าง future ที่จะเสร็จหลังจาก 2 วินาที
    let delay = time::sleep(Duration::from_secs(2));

    // สร้าง future ที่จะเสร็จทันทีด้วยค่า "hello"
    let message = async { "hello" };

    // ใช้ select! เพื่อรอให้ใดๆ ของสอง future นี้เสร็จก่อน
    tokio::select! {
        msg = message => {
            println!("Received message: {}", msg);
        }
        _ = delay => {
            println!("Time is up!");
        }
    }
}

// สมมติว่าเรามี ws_stream และ tcp_stream
tokio::select! {
    // รับข้อความจาก WebSocket และส่งไปยัง TCP
    result = ws_stream.next() => {
        match result {
            Some(Ok(msg)) => {
                if let Err(e) = tcp_stream.write_all(&msg.into()).await {
                    eprintln!("Failed to write to TCP: {}", e);
                    break;
                }
            }
            Some(Err(e)) => {
                eprintln!("WebSocket error: {}", e);
                break;
            }
            None => {
                // WebSocket ปิดการเชื่อมต่อ
                break;
            }
        }
    }
    // รับข้อมูลจาก TCP และส่งไปยัง WebSocket
    result = tcp_stream.read(&mut buf) => {
        match result {
            Ok(0) => {
                // TCP ปิดการเชื่อมต่อ
                break;
            }
            Ok(n) => {
                if let Err(e) = ws_stream.send(web_socket::Message::Binary(
                    buf[..n].to_vec()
                )).await {
                    eprintln!("Failed to send to WebSocket: {}", e);
                    break;
                }
            }
            Err(e) => {
                eprintln!("Failed to read from TCP: {}", e);
                break;
            }
        }
    }
}

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::signal;

#[tokio::main]
async fn main() -> io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Server listening on 127.0.0.1:8080");

    // สร้าง future ที่จะเสร็จเมื่อได้รับสัญญาณ Ctrl+C
    let shutdown_signal = async {
        signal::ctrl_c().await.expect("Failed to install CTRL+C handler");
        println!("Received shutdown signal, shutting down...");
    };

    // เก็บตัวจัดการ task ทั้งหมดเพื่อที่เราจะสามารถรอให้พวกมันเสร็จสิ้นได้เมื่อปิด
    let mut tasks = Vec::new();

    loop {
        tokio::select! {
            // รอการเชื่อมต่อใหม่
            result = listener.accept() => {
                match result {
                    Ok((socket, addr)) => {
                        println!("Accepted connection from: {}", addr);
                        // สร้าง task ใหม่สำหรับการเชื่อมต่อนี้
                        let handle = tokio::spawn(process_socket(socket));
                        tasks.push(handle);
                    }
                    Err(e) => {
                        eprintln!("Failed to accept connection: {}", e);
                    }
                }
            }
            // รอสัญญาณปิดการทำงาน
            _ = shutdown_signal => {
                println!("Shutdown signal received, stopping acceptance of new connections.");
                break; // ออกจากลูปการรับการเชื่อมต่อใหม่
            }
        }
    }

    // รอให้ทุก task ที่กำลังทำงานเสร็จสิ้น
    println!("Waiting for {} active connections to finish...", tasks.len());
    for task in tasks {
        let _ = task.await;
    }

    println!("All connections closed. Goodbye!");
    Ok(())
}

async fn process_socket(mut socket: TcpStream) -> io::Result<()> {
    let mut buffer = [0; 1024];

    loop {
        let n = match socket.read(&mut buffer).await {
            Ok(0) => return Ok(()), // Client ปิดการเชื่อมต่อ
            Ok(n) => n,
            Err(e) => return Err(e),
        };

        if let Err(e) = socket.write_all(&buffer[..n]).await {
            return Err(e);
        }
    }
}

async fn handle_connection(ws_stream: WebSocketStream<MaybeTlsStream<TcpStream>>, addr: SocketAddr) {
    // แยกการเชื่อมต่อ WebSocket ออกเป็นตัวรับและตัวส่ง
    let (mut ws_sender, mut ws_receiver) = ws_stream.split();

    // เชื่อมต่อไปยัง TCP server ตามที่ตั้งค่าไว้
    let tcp_stream = TcpStream::connect(&config.server.target_addr).await?;
    let (mut tcp_reader, mut tcp_writer) = io::split(tcp_stream);

    // สร้างช่องทางสำหรับส่งสัญญาณยกเลิกระหว่างสองทิศทาง
    let (tx_to_ws, mut rx_to_ws) = mpsc::unbounded_channel();
    let (tx_to_tcp, mut rx_to_tcp) = mpsc::unbounded_channel();

    // งานที่ 1: อ่านจาก WebSocket และส่งไปยัง TCP
    let ws_to_tcp_task = tokio::spawn(async move {
        while let Some(msg) = ws_receiver.next().await {
            match msg {
                Ok(Message::Binary(data)) => {
                    if tx_to_tcp.send(data).is_err() {
                        break; // ตัวรับถูกยกเลิกแล้ว
                    }
                }
                Ok(Message::Close(_)) => {
                    // ส่งสัญญาณปิดไปยัง TCP ฝั่งผู้ส่ง
                    let _ = tx_to_tcp.send(Vec::new()).await;
                    break;
                }
                Err(e) => {
                    eprintln!("WebSocket error: {}", e);
                    break;
                }
                _ => {} // ละเลยข้อความประเภทอื่นๆ เช่น Ping, Pong, Text
            }
        }
    });

    // งานที่ 2: อ่านจาก TCP และส่งไปยัง WebSocket
    let tcp_to_ws_task = tokio::spawn(async move {
        let mut buffer = vec![0; 4096];
        loop {
            let n = match tcp_reader.read(&mut buffer).await {
                Ok(0) => break, // TCP ปิดการเชื่อมต่อ
                Ok(n) => n,
                Err(e) => {
                    eprintln!("TCP read error: {}", e);
                    break;
                }
            };

            // ส่งข้อมูลไปยัง WebSocket ผ่าน channel เพื่อหลีกเลี่ยงการล็อกบน ws_sender
            if tx_to_ws.send(buffer[..n].to_vec()).is_err() {
                break; // ตัวรับถูกยกเลิกแล้ว
            }
        }
    });

    // งานที่ 3: ส่งจาก channel ไปยัง WebSocket
    let ws_sender_task = tokio::spawn(async move {
        while let Some(data) = rx_to_ws.recv().await {
            if let Err(e) = ws_sender.send(Message::Binary(data)).await {
                eprintln!("WebSocket write error: {}", e);
                break;
            }
        }
    });

    // งานที่ 4: ส่งจาก channel ไปยัง TCP
    let tcp_sender_task = tokio::spawn(async move {
        while let Some(data) = rx_to_tcp.recv().await {
            if data.is_empty() {
                // สัญญาณปิดการเชื่อมต่อ
                break;
            }
            if let Err(e) = tcp_writer.write_all(&data).await {
                eprintln!("TCP write error: {}", e);
                break;
            }
        }
    });

    // รอให้งานทั้งสี่เสร็จสิ้น หรือเกิดข้อผิดพลาดขึ้น
    tokio::select! {
        _ = ws_to_tcp_task => {}
        _ = tcp_to_ws_task => {}
        _ = ws_sender_task => {}
        _ = tcp_sender_task => {}
    }

    // ปิดการเชื่อมต่อทั้งสองฝั่งอย่างสงบ
    let _ = ws_sender.close().await;
    let _ = tcp_writer.shutdown().await;
}

async fn run_server(config: Config) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("0.0.0.0:{}", config.server.port);
    let listener = TcpListener::bind(&addr).await?;
    println!("WebSocket server listening on {}", addr);

    // สร้าง future ที่จะเสร็จเมื่อได้รับสัญญาณ Ctrl+C
    let shutdown_signal = async {
        signal::ctrl_c().await.expect("Failed to install CTRL+C handler");
        println!("\nReceived shutdown signal, shutting down gracefully...");
    };

    // เก็บตัวจัดการ task ของการเชื่อมต่อทั้งหมด
    let mut connection_tasks = Vec::new();

    // ลูปหลักสำหรับการรับการเชื่อมต่อใหม่
    loop {
        tokio::select! {
            // รอการเชื่อมต่อใหม่
            result = listener.accept() => {
                match result {
                    Ok((ws_stream, addr)) => {
                        println!("New WebSocket connection from: {}", addr);
                        // สร้าง task ใหม่สำหรับการเชื่อมต่อนี้
                        let handle = tokio::spawn(handle_connection(ws_stream, addr, config.clone()));
                        connection_tasks.push(handle);
                    }
                    Err(e) => {
                        eprintln!("Failed to accept connection: {}", e);
                    }
                }
            }
            // รอสัญญาณปิดการทำงาน
            _ = shutdown_signal => {
                println!("Shutdown signal received, stopping acceptance of new connections.");
                break; // ออกจากลูปการรับการเชื่อมต่อใหม่
            }
        }
    }

    // รอให้ทุกการเชื่อมต่อที่ยังทำงานอยู่เสร็จสิ้น
    println!("Waiting for {} active connections to close...", connection_tasks.len());
    for task in connection_tasks {
        let _ = task.await;
    }

    println!("All connections closed. Server shut down gracefully.");
    Ok(())
}