//! An async notification channel.

//!

//! This channel allows one task to notify another. No data is passed from the sender to receiver. A

//! [`NotifySender`] may send multiple notifications and a [`NotifyReceiver`] may receive multiple

//! notifications. Notifications will be coalesced, so if a `NotifySender` sends multiple

//! notifications, the `NotifyReceiver` may or may not receive all of the notifications. If there

//! are multiple `NotifyReceiver`s, each will be notified.

//!

//! An optional type can be attached to the `NotifySender` and `NotifyReceiver` to identify the

//! purpose of the notifications and to provide type checking.

// TODO(arti#534): we expect to use this for flow control, so we should remove this later

#![cfg_attr(not(test), expect(dead_code))]

use std::marker::PhantomData;

use std::pin::Pin;

use std::task::{Context, Poll};

use educe::Educe;

use futures::stream::{Fuse, FusedStream};

use futures::{Stream, StreamExt};

use pin_project::pin_project;

use postage::watch;

/// A [`NotifySender`] which can notify [`NotifyReceiver`]s.

///

/// See the [module documentation](self) for details.

#[derive(Educe)]

#[educe(Debug)]

pub(crate) struct NotifySender<T = ()> {

    /// The "sender" we use to implement the async behaviour.

    sender: watch::Sender<()>,

    /// Allows the user to optionally attach a type marker to identify the purpose of the

    /// notifications.

    #[educe(Debug(ignore))]

    _marker: PhantomData<fn() -> T>,

}

/// A [`NotifyReceiver`] which can receive notifications from a [`NotifySender`].

///

/// See the [module documentation](self) for details.

// We should theoretically be able to impl `Clone`, but `Fuse` does not implement `Clone` so we'd

// have to implement something manually. If we do want `Clone` in the future, be careful about the

// initial state of the new `NotifyReceiver` (see the `try_recv` in `NotifySender::subscribe`).

#[derive(Educe)]

#[educe(Debug)]

#[pin_project]

pub(crate) struct NotifyReceiver<T = ()> {

    /// The "receiver" we use to implement the async behaviour.

    #[pin]

    receiver: Fuse<watch::Receiver<()>>,

    /// Allows the user to optionally attach a type marker to identify the purpose of the

    /// notifications.

    #[educe(Debug(ignore))]

    _marker: PhantomData<fn() -> T>,

}

impl NotifySender {

    /// Create a new untyped [`NotifySender`].

}

impl<T> NotifySender<T> {

    /// Create a new typed [`NotifySender<T>`].

    /// Notify all [`NotifyReceiver`]s.

        // from `postage::watch::Sender`:

        // > Mutably borrows the contained value, blocking the channel while the borrow is held.

        // > After the borrow is released, receivers will be notified of a new value.

    /// Create a new [`NotifyReceiver`] for this [`NotifySender`].

    ///

    /// A new `NotifyReceiver` will not see any past notifications.

        // a `watch::Receiver` will always return the existing status of the `watch::Sender` as the

        // first stream item, so we need to recv and discard it so that this `NotifyReceiver` begins

        // in the "pending" state

        use postage::stream::Stream as PostageStream;

        use postage::stream::TryRecvError;

            Err(TryRecvError::Pending),

        );

}

impl<T> Stream for NotifyReceiver<T> {

    type Item = ();

}

// the `NotifyReceiver` stores a `Fuse`

impl<T> FusedStream for NotifyReceiver<T> {

}

#[cfg(test)]

mod test {

    #![allow(clippy::unwrap_used)]

    use super::*;

    use futures::FutureExt;

    #[test]

    fn notify() {

        tor_rtmock::MockRuntime::test_with_various(|_rt| async move {

            let mut sender = NotifySender::new();

            let mut receiver = sender.subscribe();

            // receivers should initially wait for a notification

            assert_eq!(receiver.next().now_or_never(), None);

            assert_eq!(receiver.next().now_or_never(), None);

            sender.notify();

            // we should receive a single notification

            assert_eq!(receiver.next().now_or_never(), Some(Some(())));

            assert_eq!(receiver.next().now_or_never(), None);

            sender.notify();

            sender.notify();

            sender.notify();

            // we should still receive a single notification

            assert_eq!(receiver.next().now_or_never(), Some(Some(())));

            assert_eq!(receiver.next().now_or_never(), None);

            sender.notify();

            drop(sender);

            // we should see the last notification, and then since we dropped the sender, the stream

            // should indicate that it's finished

            assert_eq!(receiver.next().now_or_never(), Some(Some(())));

            assert_eq!(receiver.next().now_or_never(), Some(None));

            assert_eq!(receiver.next().now_or_never(), Some(None));

        });

    }

    #[test]

    fn notify_multiple_receivers() {

        tor_rtmock::MockRuntime::test_with_various(|_rt| async move {

            let mut sender = NotifySender::new();

            let mut receiver_1 = sender.subscribe();

            let mut receiver_2 = sender.subscribe();

            sender.notify();

            let mut receiver_3 = sender.subscribe();

            // first two receivers should each receive a notification

            assert_eq!(receiver_1.next().now_or_never(), Some(Some(())));

            assert_eq!(receiver_2.next().now_or_never(), Some(Some(())));

            // third receiver should not receive a notification since it was created after the

            // notification was generated

            assert_eq!(receiver_3.next().now_or_never(), None);

        });

    }

}