use qanat::mpsc::{channel, RecvError, TryRecvError, SendError, TrySendError, restore_channel_from_snapshot, unbounded_channel};

use futures_lite::future::block_on;
use easy_parallel::Parallel;

#[test]
fn basic_unbound_mpsc() {
    let (s, mut r1) = unbounded_channel();
    let s1 = s.clone();

    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));

    s.send(7).unwrap();
    assert_eq!(r1.try_recv().unwrap(), 7);
    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
    drop(s1);
    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));

    s.send(8).unwrap();
    assert_eq!(r1.try_recv().unwrap(), 8);
    drop(s);
    assert_eq!(r1.try_recv(), Err(TryRecvError::Closed));
}

#[test]
fn sync_mpsc() {
    let (s, mut r1) = channel(1.try_into().unwrap());
    let s1 = s.clone();

    s.try_send(7).unwrap();
    assert_eq!(s.try_send(7), Err((TrySendError::Full, 7)));
    
    assert_eq!(s1.try_send(7), Err((TrySendError::Full, 7)));
    drop(s1);

    assert_eq!(r1.try_recv().unwrap(), 7);
    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));

    s.try_send(8).unwrap();
    assert_eq!(r1.try_recv().unwrap(), 8);
    drop(s);
    assert_eq!(r1.try_recv(), Err(TryRecvError::Closed));
}

#[test]
fn async_mpsc() {
    block_on(async {
        let (s, mut r1) = channel(1.try_into().unwrap());
        let s1 = s.clone();

        s.send(7).await.unwrap();
        assert_eq!(s.try_send(7), Err((TrySendError::Full, 7)));

        assert_eq!(s1.try_send(7), Err((TrySendError::Full, 7)));
        drop(s1);

        assert_eq!(r1.recv().await.unwrap(), 7);
        assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));

        s.send(8).await.unwrap();
        assert_eq!(r1.recv().await.unwrap(), 8);
        drop(s);
        assert_eq!(r1.recv().await, Err(RecvError::Closed));
    });
}

#[test]
fn parallel_mpsc() {
    for _ in 0..1000 {
        let (s1, mut r1) = channel(2.try_into().unwrap());
        let s2 = s1.clone();

        Parallel::new()
        .add(move || {
            block_on(async {
                let mut n = 0;
                loop {
                    if n == 1000 {
                        break;
                    }
                    let ret = r1.recv().await.unwrap();
                    assert_eq!(ret, n);
                    n += 1;
                };
                assert_eq!(r1.recv().await.unwrap(), 1);
                drop(r1);
            })
        })
        .add(move || {
            block_on(async {
                for n in 0..1000 {
                    if n % 2 == 0 {
                        s1.send(n).await.unwrap();
                    } else {
                        s2.send(n).await.unwrap();
                    }
                }
                drop(s1);
                // First send should pass
                s2.send(1).await.unwrap();
                // 2nd and 3 may fail
                match s2.send(2).await {
                    Ok(_) => (),
                    Err(e) => assert_eq!(e, (SendError::Closed, 2))
                }
                match s2.send(3).await {
                    Ok(_) => (),
                    Err(e) => assert_eq!(e, (SendError::Closed, 3))
                }
                // Last one must fail because we will be waiting for reader
                assert_eq!(s2.send(4).await, Err((SendError::Closed, 4)));
            })
        })
        .run();
    }
}

#[test]
fn snapshot_mpsc() {
    let (s1, mut r1) = channel(2.try_into().unwrap());
    let s2 = s1.clone();

    s1.try_send(1).unwrap();
    s2.try_send(2).unwrap();

    r1.try_recv().unwrap();

    s1.try_send(3).unwrap();
    assert_eq!(s1.try_send(4), Err((TrySendError::Full, 4)));

    let snapshot = r1.snapshot();

    drop(s1);
    drop(s2);
    drop(r1);

    let (s1, mut r1) = restore_channel_from_snapshot(snapshot)
        .unwrap();
    let s2 = s1.clone();

    assert_eq!(s1.len(), 2);
    
    assert_eq!(s2.try_send(4), Err((TrySendError::Full, 4)));

    assert_eq!(r1.try_recv(), Ok(2));
    assert_eq!(r1.len(), 1);

    assert_eq!(s2.try_send(4), Ok(()));

    assert_eq!(r1.try_recv(), Ok(3));
    assert_eq!(r1.try_recv(), Ok(4));
    assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
}

#[test]
fn same_channel_mpsc() {
    let (s1, r1) = channel::<usize>(5.try_into().unwrap());
    let (s2, r2) = channel::<usize>(5.try_into().unwrap());
    let (s3, r3) = channel::<usize>(2.try_into().unwrap());

    assert!(s1.same_channel(&r1));
    assert!(!r1.same_channel(&s2));
    assert!(!s3.same_channel(&s1));
    assert!(!s1.same_channel(&s3));
    assert!(r2.same_channel(&s2.clone()));
    assert!(r3.same_channel(&s3.clone()));
}

#[test]
fn same_channel_unbound_mpsc() {
    let (s1, r1) = unbounded_channel::<usize>();
    let (s2, r2) = unbounded_channel::<usize>();
    let (s3, r3) = unbounded_channel::<usize>();

    assert!(s1.same_channel(&r1));
    assert!(!r1.same_channel(&s2));
    assert!(!s3.same_channel(&s1));
    assert!(!s1.same_channel(&s3));
    assert!(r2.same_channel(&s2.clone()));
    assert!(r3.same_channel(&s3.clone()));
}
