#![cfg(target_os = "linux")]

mod common;

use std::io::{Read, Write};
use std::net::{Shutdown, TcpListener as StdTcpListener};
use std::ptr::NonNull;
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};

use dambi::Bytes;
use dope_core::driver::{Backend, DriverConfig};
use dope_futures::conn::{Conn, ConnState};
use dope_transport::Tcp;
use dope_uring::{CurrentSlot, UringBackend, set_current};

use common::Drive;

const TICK_BUDGET: Duration = Duration::from_secs(5);

#[test]
fn connect_send_recv_roundtrip() {
    let listener = StdTcpListener::bind("127.0.0.1:0").expect("bind tcp listener");
    let addr = listener.local_addr().expect("local addr");

    let (server_tx, server_rx) = mpsc::sync_channel::<Vec<u8>>(1);
    let server_thread = thread::spawn(move || {
        let (mut sock, _) = listener.accept().expect("accept");
        sock.set_read_timeout(Some(Duration::from_secs(3))).ok();
        let mut buf = [0u8; 64];
        let n = sock.read(&mut buf).expect("server read");
        server_tx.send(buf[..n].to_vec()).ok();
        sock.write_all(b"echo:hello").expect("server write");
        let _ = sock.shutdown(Shutdown::Write);
    });

    let driver_cfg = <UringBackend as Backend>::Config::for_tcp_profile(128, 16, 4096);
    let mut driver = <UringBackend as Backend>::new_driver(driver_cfg).expect("driver");
    let driver_ptr = NonNull::from(&mut driver);
    set_current(Some(CurrentSlot::new(driver_ptr, std::ptr::null())));

    let mut conn = Conn::<Tcp>::connect(addr, &mut driver, 0).expect("connect");

    Drive::until(&mut conn, &mut driver, TICK_BUDGET, |c| {
        matches!(c.state(), ConnState::Active)
    });

    conn.try_send(Bytes::from_static(b"hello"))
        .expect("try_send");

    let mut received: Vec<u8> = Vec::new();
    Drive::until(&mut conn, &mut driver, TICK_BUDGET, |c| {
        while let Some(chunk) = c.try_recv() {
            received.extend_from_slice(&*chunk);
        }
        received.len() >= b"echo:hello".len()
    });
    assert_eq!(received, b"echo:hello");

    let server_received = server_rx
        .recv_timeout(Duration::from_secs(2))
        .expect("server msg");
    assert_eq!(server_received, b"hello");

    drop(conn);
    set_current(None);
    server_thread.join().ok();
}

#[test]
fn connect_to_nonlistening_port_errors() {
    let listener = StdTcpListener::bind("127.0.0.1:0").expect("bind probe");
    let addr = listener.local_addr().expect("local addr");
    drop(listener);

    let driver_cfg = <UringBackend as Backend>::Config::for_tcp_profile(64, 16, 4096);
    let mut driver = <UringBackend as Backend>::new_driver(driver_cfg).expect("driver");
    let driver_ptr = NonNull::from(&mut driver);
    set_current(Some(CurrentSlot::new(driver_ptr, std::ptr::null())));

    let mut conn = Conn::<Tcp>::connect(addr, &mut driver, 0).expect("connect submitted");

    Drive::until(&mut conn, &mut driver, Duration::from_secs(2), |c| {
        matches!(c.state(), ConnState::Error)
    });
    assert!(conn.last_error().is_some());

    drop(conn);
    set_current(None);
}
