// Part of this is courtesy of the smol-rs team
// https://github.com/smol-rs/async-broadcast/blob/master/tests/test.rs
use qanat::broadcast::{channel, RecvError, TryRecvError, restore_channel_from_snapshot};

use std::sync::mpsc;
use futures_lite::future::block_on;
use easy_parallel::Parallel;

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

    s.send(7);
    assert_eq!(r1.try_recv().unwrap(), 7);
    assert_eq!(r2.try_recv().unwrap(), 7);

    let mut r3 = r1.clone();
    s.send(8);
    assert_eq!(r1.try_recv().unwrap(), 8);
    assert_eq!(r2.try_recv().unwrap(), 8);
    assert_eq!(r3.try_recv().unwrap(), 8);
}

#[test]
fn basic_async_broadcast() {
    block_on(async {
        let (mut s, mut r1) = channel(10.try_into().unwrap());
        let mut r2 = r1.clone();

        s.send(7);
        assert_eq!(r1.recv().await.unwrap(), 7);
        assert_eq!(r2.recv().await.unwrap(), 7);

        let mut r3 = r1.clone();
        s.send(8);
        assert_eq!(r1.recv().await.unwrap(), 8);
        assert_eq!(r2.recv().await.unwrap(), 8);
        assert_eq!(r3.recv().await.unwrap(), 8);
    });
}

#[test]
fn exceptions_broadcast() {
    block_on(async {
        let (mut s, mut r1) = channel(10.try_into().unwrap());
        let mut r2 = r1.clone();

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

        let v: Vec<u8> = Vec::new();
        s.send(v);

        assert_eq!(r1.recv().await.unwrap().len(), 0);
        assert_eq!(r2.recv().await.unwrap().len(), 0);

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

        let mut r3 = r1.clone();
        assert_eq!(r3.try_recv(), Err(TryRecvError::Empty));

        s.send(vec![1u8]);
        assert_eq!(r1.recv().await.unwrap().len(), 1);
        assert_eq!(r2.recv().await.unwrap().len(), 1);
        assert_eq!(r3.recv().await.unwrap().len(), 1);
        
        let s2 = s.clone();
        drop(s);

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

        drop(s2);

        assert_eq!(r1.recv().await, Err(RecvError::Closed));
        assert_eq!(r2.recv().await, Err(RecvError::Closed));
        assert_eq!(r3.recv().await, Err(RecvError::Closed));
    });
}

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

    let (sender_sync_send, sender_sync_recv) = mpsc::channel();
    let (receiver_sync_send, receiver_sync_recv) = mpsc::channel();

    Parallel::new()
    .add(move || {
        sender_sync_recv.recv().unwrap();

        s1.send(7);
        s2.send(8);
        receiver_sync_send.send(()).unwrap();

        drop(s1);
        drop(s2);
        receiver_sync_send.send(()).unwrap();
    })
    .add(move || {
        assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
        assert_eq!(r2.try_recv(), Err(TryRecvError::Empty));
        sender_sync_send.send(()).unwrap();

        receiver_sync_recv.recv().unwrap();
        assert_eq!(r1.try_recv().unwrap(), 7);
        assert_eq!(r1.try_recv().unwrap(), 8);
        assert_eq!(r2.try_recv().unwrap(), 7);
        assert_eq!(r2.try_recv().unwrap(), 8);

        receiver_sync_recv.recv().unwrap();
        assert_eq!(r1.try_recv(), Err(TryRecvError::Closed));
        assert_eq!(r2.try_recv(), Err(TryRecvError::Closed));
    })
    .run();
}

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

    let (sender_sync_send, sender_sync_recv) = mpsc::channel();
    let (receiver_sync_send, receiver_sync_recv) = mpsc::channel();

    Parallel::new()
    .add(move || {
        sender_sync_recv.recv().unwrap();

        s1.send(7);
        s2.send(8);
        receiver_sync_send.send(()).unwrap();
        sender_sync_recv.recv().unwrap();

        s1.send(9);
        s1.send(10);
        s1.send(11);

        drop(s1);
        drop(s2);
        receiver_sync_send.send(()).unwrap();
    })
    .add(move || {
        block_on(async move {
            assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
            assert_eq!(r2.try_recv(), Err(TryRecvError::Empty));
            sender_sync_send.send(()).unwrap();

            receiver_sync_recv.recv().unwrap();
            assert_eq!(r1.recv().await.unwrap(), 7);
            assert_eq!(r1.recv().await.unwrap(), 8);
            assert_eq!(r2.recv().await.unwrap(), 7);
            assert_eq!(r2.recv().await.unwrap(), 8);

            sender_sync_send.send(()).unwrap();
            receiver_sync_recv.recv().unwrap();

            assert_eq!(r1.recv().await, Err(RecvError::Lagged));
            assert_eq!(r2.recv().await, Err(RecvError::Lagged));

            assert_eq!(r1.recv().await.unwrap(), 10);
            assert_eq!(r1.recv().await.unwrap(), 11);
            assert_eq!(r2.try_recv().unwrap(), 10);
            assert_eq!(r2.try_recv().unwrap(), 11);

            assert_eq!(r1.recv().await, Err(RecvError::Closed));
            assert_eq!(r2.recv().await, Err(RecvError::Closed));
        })
    })
    .run();

}

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

    s.send(7);

    assert_eq!(s.capacity(), 1);
    assert_eq!(s.len(), 1);

    s.set_capacity(1.try_into().unwrap());

    assert_eq!(r1.try_recv().unwrap(), 7);

    assert_eq!(s.capacity(), 1);
    assert_eq!(s.len(), 1);

    s.set_capacity(2.try_into().unwrap());

    assert_eq!(s.capacity(), 2);
    assert_eq!(s.len(), 1);

    let mut r3 = r1.clone();
    
    s.send(8);

    assert_eq!(s.len(), 2);

    assert_eq!(r2.try_recv().unwrap(), 7);

    assert_eq!(r1.try_recv().unwrap(), 8);
    assert_eq!(r2.try_recv().unwrap(), 8);

    s.send(9);

    s.set_capacity(1.try_into().unwrap());

    assert_eq!(s.capacity(), 1);
    assert_eq!(s.len(), 1);

    assert_eq!(r1.try_recv().unwrap(), 9);
    assert_eq!(r2.try_recv().unwrap(), 9);
    
    assert_eq!(r3.try_recv(), Err(TryRecvError::Lagged));
    assert_eq!(r3.try_recv().unwrap(), 9);
}

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

    s1.send(1);
    s2.send(2);

    r1.try_recv().unwrap();
    r1.try_recv().unwrap();
    r2.try_recv().unwrap();

    s1.send(3);
    s1.send(4);
    s2.send(5);
    s2.send(6);
    s2.send(7);

    let snapshot = s1.snapshot();

    drop(s1);
    drop(s2);

    let resume_r1 = r1.read_position();
    let resume_r2 = r2.read_position();

    drop(r1);
    drop(r2);

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

    r1.restore(resume_r1);
    r2.restore(resume_r2);

    assert_eq!(s1.len(), 5);

    assert_eq!(r1.try_recv(), Ok(3));
    // Even if we did lag, after restore it is not accounted for
    assert_eq!(r2.try_recv(), Ok(3));
}

#[test]
fn same_channel_broadcast() {
    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(&r2.clone()));
    assert!(s3.same_channel(&r3.clone()));
}
