use std::time::Duration;

use dope_futures::listener::Listener;
use dope_futures::stream::Stream;
use dope_runtime::{DefaultBackend, block_on, spawn, timeout};
use dope_transport::Tcp;

#[test]
fn two_sequential_connections() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            for i in 0..2u8 {
                let (mut tcp, _) = listener.accept().await.unwrap();
                let mut buf = [0u8; 1];
                tcp.read_exact(&mut buf).await.unwrap();
                assert_eq!(buf[0], b'A' + i);
                tcp.write_all(&buf).await.unwrap();
            }
        });
        for i in 0..2u8 {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            tcp.write_all(&[b'A' + i]).await.unwrap();
            let mut out = [0u8; 1];
            timeout(Duration::from_secs(2), async {
                tcp.read_exact(&mut out).await.unwrap();
            })
            .await
            .expect("read timeout");
            assert_eq!(out[0], b'A' + i);
        }
        timeout(Duration::from_secs(2), server)
            .await
            .expect("server timeout");
    });
}

fn run_sequential(n: usize) {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            for i in 0..n {
                let (mut tcp, _) = listener.accept().await.unwrap();
                let mut buf = [0u8; 4];
                tcp.read_exact(&mut buf).await.unwrap();
                assert_eq!(buf, [i as u8; 4]);
                tcp.write_all(&buf).await.unwrap();
            }
        });
        for i in 0..n {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            let msg = [i as u8; 4];
            tcp.write_all(&msg).await.unwrap();
            let mut out = [0u8; 4];
            timeout(Duration::from_secs(5), async {
                tcp.read_exact(&mut out).await.unwrap();
            })
            .await
            .expect("client read timeout");
            assert_eq!(out, msg);
        }
        timeout(Duration::from_secs(5), server)
            .await
            .expect("server timeout");
    });
}

#[test]
fn sequential_4() {
    run_sequential(4);
}
#[test]
fn sequential_8() {
    run_sequential(8);
}
#[test]
fn sequential_16() {
    run_sequential(16);
}

fn run_concurrent_via_spawn(n: usize) {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            for _ in 0..n {
                let (mut tcp, _) = listener.accept().await.unwrap();
                spawn(async move {
                    let mut buf = [0u8; 4];
                    tcp.read_exact(&mut buf).await.unwrap();
                    tcp.write_all(&buf).await.unwrap();
                });
            }
        });
        for i in 0..n {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            let msg = [i as u8; 4];
            tcp.write_all(&msg).await.unwrap();
            let mut out = [0u8; 4];
            timeout(Duration::from_secs(5), async {
                tcp.read_exact(&mut out).await.unwrap();
            })
            .await
            .expect("client read timeout");
            assert_eq!(out, msg);
        }
        timeout(Duration::from_secs(5), server)
            .await
            .expect("server timeout");
    });
}

#[test]
fn concurrent_4() {
    run_concurrent_via_spawn(4);
}
#[test]
fn concurrent_5() {
    run_concurrent_via_spawn(5);
}
#[test]
fn concurrent_6() {
    run_concurrent_via_spawn(6);
}
#[test]
fn concurrent_8() {
    run_concurrent_via_spawn(8);
}
#[test]
fn concurrent_16() {
    run_concurrent_via_spawn(16);
}

#[test]
fn sixteen_concurrent_via_spawn() {
    const N: usize = 16;
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            for _ in 0..N {
                let (mut tcp, _) = listener.accept().await.unwrap();
                spawn(async move {
                    let mut buf = [0u8; 4];
                    tcp.read_exact(&mut buf).await.unwrap();
                    tcp.write_all(&buf).await.unwrap();
                });
            }
        });
        for i in 0..N {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            let msg = [i as u8; 4];
            tcp.write_all(&msg).await.unwrap();
            let mut out = [0u8; 4];
            timeout(Duration::from_secs(5), async {
                tcp.read_exact(&mut out).await.unwrap();
            })
            .await
            .expect("client read timeout");
            assert_eq!(out, msg);
        }
        timeout(Duration::from_secs(5), server)
            .await
            .expect("server timeout");
    });
}

#[test]
fn three_concurrent_via_spawn() {
    const N: usize = 3;
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            for _ in 0..N {
                let (mut tcp, _) = listener.accept().await.unwrap();
                spawn(async move {
                    let mut buf = [0u8; 4];
                    tcp.read_exact(&mut buf).await.unwrap();
                    tcp.write_all(&buf).await.unwrap();
                });
            }
        });
        for i in 0..N {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            let msg = [i as u8; 4];
            tcp.write_all(&msg).await.unwrap();
            let mut out = [0u8; 4];
            timeout(Duration::from_secs(2), async {
                tcp.read_exact(&mut out).await.unwrap();
            })
            .await
            .expect("client read timeout");
            assert_eq!(out, msg);
        }
        timeout(Duration::from_secs(2), server)
            .await
            .expect("server timeout");
    });
}

const T: Duration = Duration::from_secs(2);

#[test]
fn one_byte_one_way_client_to_server() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            let mut buf = [0u8; 1];
            tcp.read_exact(&mut buf).await.unwrap();
            assert_eq!(buf[0], b'A');
        });
        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        tcp.write_all(b"A").await.unwrap();
        timeout(T, server).await.expect("server timed out");
    });
}

#[test]
fn one_byte_one_way_server_to_client() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            tcp.write_all(b"B").await.unwrap();
        });
        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        let mut buf = [0u8; 1];
        timeout(T, async {
            tcp.read_exact(&mut buf).await.unwrap();
        })
        .await
        .expect("read timed out");
        assert_eq!(buf[0], b'B');
        server.await;
    });
}

#[test]
fn one_byte_roundtrip() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    block_on(async move {
        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            let mut buf = [0u8; 1];
            tcp.read_exact(&mut buf).await.unwrap();
            tcp.write_all(&buf).await.unwrap();
        });
        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        tcp.write_all(b"X").await.unwrap();
        let mut out = [0u8; 1];
        timeout(T, async {
            tcp.read_exact(&mut out).await.unwrap();
        })
        .await
        .expect("client read timed out");
        assert_eq!(out[0], b'X');
        timeout(T, server).await.expect("server timed out");
    });
}
