#![forbid(unsafe_code)]

use core::future::Future;
use core::pin::Pin;
use core::task::{Context, Poll};
use std::sync::{Arc, Mutex};
use std::task::Waker;

enum State<T> {
    Unset,
    Set(Option<T>),
}

struct Inner<T> {
    value: State<T>,
    waker: Option<Waker>,
}

/// A Future that resolves when you call [`set`](#method.set) on its clone.
///
/// Uses [`Arc`](https://doc.rust-lang.org/stable/std/sync/struct.Arc.html)
/// internally.
/// All clones of this struct use the same inner struct on the heap.
///
/// You can call `set` on any clone of this struct.
///
/// You can call `set` once and `await` once, obtaining the value.
///
/// Subsequent or concurrent awaits of the struct will wait forever.
pub struct Promise<T: Send + 'static> {
    inner: Arc<Mutex<Inner<T>>>,
}

impl<T: Send + 'static> Promise<T> {
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                value: State::Unset,
                waker: None,
            })),
        }
    }

    /// Returns true if [`set`](#method.set) has been called.
    #[must_use]
    pub fn is_set(&self) -> bool {
        !matches!(self.inner.lock().unwrap().value, State::Unset)
    }

    /// Sets the value of the struct and wakes up any awaiting task.
    ///
    /// Consumes the struct.
    ///
    /// Panics if the value was already set.
    pub fn set(self, value: T) {
        if self.try_set(value).is_err() {
            panic!("value already set");
        }
    }

    /// Sets the value of the struct and wakes up any awaiting task.
    ///
    /// # Errors
    /// Returns `Err(value)` if the value was already set.
    /// This returns the value just passed in.
    pub fn try_set(&self, value: T) -> Result<(), T> {
        let mut guard = self.inner.lock().unwrap();
        if let State::Unset = guard.value {
            guard.value = State::Set(Some(value));
            if let Some(waker) = guard.waker.take() {
                waker.wake();
            }
            Ok(())
        } else {
            Err(value)
        }
    }
}

impl<T: Send + 'static> Default for Promise<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T: Send + 'static> Clone for Promise<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<T: Send + 'static> Future for Promise<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut guard = self.inner.lock().unwrap();
        #[allow(clippy::option_if_let_else)]
        match &mut guard.value {
            State::Set(None) => {
                // This future previously completed and is being polled again.
                Poll::Pending
            }
            State::Set(ref mut opt_value) => Poll::Ready(opt_value.take().unwrap()),
            State::Unset => {
                guard.waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::super::tests::expect_elapsed;
    use super::*;
    use core::time::Duration;
    use safina_async_test::async_test;
    use std::time::Instant;

    #[async_test]
    async fn is_set() {
        let promise = Promise::new();
        let clone1 = promise.clone();
        let clone2 = promise.clone();
        assert!(!promise.is_set());
        assert!(!clone1.is_set());
        assert!(!clone2.is_set());
        promise.set(true);
        assert!(clone1.is_set());
        assert!(clone2.is_set());
        assert_eq!(true, clone1.await);
        assert!(clone2.is_set());
    }

    #[test]
    fn test_set() {
        let promise = Promise::new();
        let promise_clone = promise.clone();
        promise.set(true);
        std::panic::catch_unwind(|| promise_clone.set(true)).unwrap_err();
    }

    #[test]
    #[allow(clippy::let_unit_value)]
    fn test_try_set() {
        let promise = Promise::new();
        let _: () = promise.try_set(true).unwrap();
        assert_eq!(true, promise.try_set(true).unwrap_err());
    }

    #[async_test]
    async fn already_set() {
        let promise = Promise::new();
        let promise_clone = promise.clone();
        promise.set(true);
        let before = Instant::now();
        assert_eq!(true, promise_clone.await);
        expect_elapsed(before, 0..1);
    }

    #[async_test]
    async fn set_should_wake() {
        let before = Instant::now();
        let promise = Promise::new();
        let promise_clone = promise.clone();
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(100));
            promise_clone.set(true);
        });
        assert_eq!(true, promise.await);
        expect_elapsed(before, 100..200);
    }

    #[async_test]
    async fn one_one_await_returns() {
        let executor = safina_executor::Executor::default();
        let before = Instant::now();
        let promise = Promise::new();
        let receiver = {
            let (sender, receiver) = std::sync::mpsc::channel::<usize>();
            for _ in 0..5 {
                let sender_clone = sender.clone();
                let promise_clone = promise.clone();
                executor.spawn(async move {
                    safina_timer::sleep_for(Duration::from_millis(100)).await;
                    match safina_timer::with_timeout(promise_clone, Duration::from_millis(100))
                        .await
                    {
                        Ok(true) => sender_clone.send(1_usize).unwrap(),
                        Ok(false) => unreachable!(),
                        Err(safina_timer::DeadlineExceeded) => sender_clone.send(0_usize).unwrap(),
                    }
                });
            }
            receiver
        };
        promise.set(true);
        assert_eq!(1_usize, receiver.iter().sum());
        expect_elapsed(before, 0..300);
    }

    #[async_test]
    async fn repoll_awaits() {
        let promise = Promise::new();
        let promise_clone1 = promise.clone();
        let promise_clone2 = promise.clone();
        promise.set(true);
        assert_eq!(true, promise_clone1.await);
        let before = Instant::now();
        safina_timer::with_timeout(promise_clone2, Duration::from_millis(100))
            .await
            .unwrap_err();
        expect_elapsed(before, 100..200);
    }
}
