use std::cell::Cell;
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};

use dambi::{Bytes, BytesMut};
use dope_futures::AsyncRead;
use dope_futures::listener::Listener;
use dope_futures::stream::Stream;
use dope_runtime::{DefaultBackend, block_on, spawn};
use dope_transport::Tcp;

#[test]
fn connect_failure_does_not_leak_fd() {
    block_on(async {
        let result = Stream::<Tcp, DefaultBackend>::connect_with("127.0.0.1:1").await;
        assert!(result.is_err());
    });
}

#[test]
fn write_all_empty_does_not_block() {
    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 _ = listener.accept().await.unwrap();
        });

        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        tcp.write_all(&[]).await.unwrap();

        server.await;
    });
}

#[test]
fn read_buf_appends_into_bytes_mut() {
    let payload = b"read-buf-fast-path";
    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 (res, _) = tcp.write_all_owned(Bytes::from_static(payload)).await;
            res.unwrap();
        });

        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        let mut out = BytesMut::with_capacity(4);
        let n = tcp.read_buf(&mut out).await.unwrap();
        assert_eq!(n, payload.len());
        assert_eq!(&out[..], payload);

        server.await;
    });
}

struct BusyTask {
    remaining: Rc<Cell<u32>>,
}

impl Future for BusyTask {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let n = self.remaining.get();
        if n == 0 {
            return Poll::Ready(());
        }
        self.remaining.set(n - 1);
        cx.waker().wake_by_ref();
        Poll::Pending
    }
}

#[test]
fn io_not_starved_by_busy_tasks() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();

    let busy = Rc::new(Cell::new(10_000u32));
    let busy_task = busy.clone();

    block_on(async move {
        spawn(BusyTask {
            remaining: busy_task,
        });

        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            let (res, _) = tcp.write_all_owned(Bytes::from_static(b"io-works")).await;
            res.unwrap();
        });

        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        let (res, buf) = tcp.read(vec![0u8; 64]).await;
        let n = res.unwrap();
        assert_eq!(&buf[..n], b"io-works");

        server.await;
    });

    assert!(busy.get() < 10_000);
}
