use async_io::{Async, Timer};
use futures_lite::{AsyncReadExt, AsyncWriteExt};
use safe_executor::Executor;
use std::net::TcpListener;
use std::time::Duration;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let executor = Executor::new();

    for _ in 0..4 {
        let execu2r = executor.clone();
        std::thread::spawn(move || {
            async_io::block_on(execu2r.into_runner());
        });
    }

    let runner = executor.clone().into_runner();

    let execu2r = executor.clone();
    executor.spawn(async move {
        let listener = match Async::<TcpListener>::bind(([127, 0, 0, 1], 3456)) {
            Ok(listener) => listener,
            Err(_) => return,
        };

        println!("Listening on port 3456");

        loop {
            let (mut stream, _addr) = match listener.accept().await {
                Ok(tup) => tup,
                Err(_) => return,
            };

            execu2r.spawn(async move {
                let mut buffer = [0; 1024];

                while let Ok(n) = stream.read(&mut buffer).await {
                    if n == 0 {
                        break;
                    }

                    println!("{:?}: {:?}", std::thread::current().id(), &buffer[0..n]);

                    if let Ok(s) = std::str::from_utf8(&buffer[0..n]) {
                        let mut string = s.trim().chars().rev().collect::<String>();
                        string.push('\n');

                        if stream.write_all(string.as_bytes()).await.is_err() {
                            break;
                        }
                    } else if stream.write_all(&buffer[0..n]).await.is_err() {
                        break;
                    }
                }
            });
        }
    });

    let execu2r = executor.clone();
    executor.spawn(async move {
        Timer::after(Duration::from_secs(60 * 2)).await;
        execu2r.stop();
    });

    async_io::block_on(runner);
    Ok(())
}