use core::sync::atomic::Ordering;
use std::sync::Arc;

type AppendResult = Result<(), ()>;
pub type RecvResult<T> = Result<T, Empty>;

#[derive(Debug, PartialEq, Eq)]
pub struct Empty;

mod inner;
use inner::Inner;

pub struct Sender<T> {
    inner: Arc<Inner<T>>,
}
impl<T> Sender<T> {
    pub fn clone(&self) -> Self {
        self.inner.n_senders.value.fetch_add(1, Ordering::Relaxed);
        Self {
            inner: self.inner.clone(),
        }
    }
}
impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        self.inner.n_senders.value.fetch_sub(1, Ordering::Relaxed);
    }
}
impl<T> Sender<T> {
    pub fn send(&self, obj: T) -> AppendResult {
        self.inner.push_back(obj)
    }
}
pub struct Receiver<T> {
    inner: Arc<Inner<T>>,
}
impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        self.inner.n_receivers.value.fetch_sub(1, Ordering::Relaxed);
    }
}
impl<T> Receiver<T> {
    /// Blocking
    pub fn recv(&self) -> RecvResult<T> {
        self.inner.pop_front()
    }
}

pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let inner = Arc::new(Inner::new());
    (
        Sender {
            inner: inner.clone(),
        },
        Receiver { inner },
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    use threadpool::ThreadPool;

    fn threadpool() -> ThreadPool {
        threadpool::Builder::new().num_threads(16).build()
    }

    #[test]
    fn it_works() {
        let (tx, rx) = channel();

        tx.send(42usize).unwrap();

        let r = rx.recv().unwrap();
        assert_eq!(42, r);
    }

    #[test]
    fn single_inserter() {
        let pool = threadpool();
        let (tx, rx) = channel();

        pool.execute(move || {
            for i in 0..100_000 {
                let _ = tx.send(i);
            }
        });

        for _i in 0..100_000 {
            let r = rx.recv();
            assert!(r.is_ok(), "{:?}", r);
        }
        pool.join();
    }

    #[test]
    fn multiple_inserter() {
        let n_workers = 6;
        let pool = threadpool();
        let (tx, rx) = channel();
        for _ in 0..n_workers {
            let tx = tx.clone();
            pool.execute(move || {
                for i in 0..100_000 {
                    let _ = tx.send(i);
                }
            });
        }

        pool.join();

        for _i in 0..n_workers * 100_000 {
            let r = rx.recv();
            assert!(r.is_ok(), "{:?}", r);
        }
    }

    #[test]
    fn multiple_inserter_while_pulling() {
        let pool = threadpool();
        let (tx, rx) = channel();
        for _ in 0..8 {
            let tx = tx.clone();
            pool.execute(move || {
                for i in 0..100_000 {
                    let _ = tx.send(i);
                }
            });
        }

        for _i in 0..8 * 100_000 {
            let r = rx.recv();
            assert!(r.is_ok());
        }

        pool.join();
    }
}