holly 0.5.3

use futures::{prelude::*, sync::{oneshot, mpsc}};
use void::Void;

/// Wrap an existing [`Stream`] and make it infallible and stoppable.
///
/// The stream items and errors will be lifted to [`Event`].
/// The stream can also be stopped by pulling the returned [`KillCord`]
/// which will immediately produce `None` instead of any more stream
/// items.
pub fn stoppable<I: Clone, S: Stream>(id: I, s: S)
    -> (KillCord, impl Stream<Item = Event<I, S::Item, S::Error>, Error = Void>)
{
    Stoppable::new(Infallible::new(id, s))
}

/// Wrap an existing [`Stream`] and make it infallible and closable.
///
/// The stream items and errors will be lifted to [`Event`].
/// The stream can also be closed by pulling the returned [`KillCord`].
/// The closing functionality requires support from the stream which
/// must implement [`Close`].
pub fn closable<I: Clone, S: Stream + Close>(id: I, s: S)
    -> (KillCord, impl Stream<Item = Event<I, S::Item, S::Error>, Error = Void>)
{
    let (k, s) = Closable::new(s);
    (k, Infallible::new(id, s))
}

/// A kill cord stops or closes a [`Stream`].
#[derive(Debug)]
pub struct KillCord(oneshot::Sender<()>);

/// A wrapper around a [`Stream`] which can be stopped.
#[derive(Debug)]
struct Stoppable<S> {
    stream: S,
    killcord: oneshot::Receiver<()>,
}

impl<S: Stream> Stoppable<S> {
    /// Create a new stoppable stream and return it along with the
    /// [`KillCord`] to stop the stream.
    fn new(s: S) -> (KillCord, Self) {
        let (tx, rx) = oneshot::channel();
        let stream = Stoppable { stream: s, killcord: rx };
        (KillCord(tx), stream)
    }
}

impl<S: Stream> Stream for Stoppable<S> {
    type Item = S::Item;
    type Error = S::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if let Ok(Async::NotReady) = self.killcord.poll() {
            self.stream.poll()
        } else {
            Ok(Async::Ready(None))
        }
    }
}

/// A type which can be closed.
///
/// The semantics of a closed stream are that the stream source
/// will be closed, but items which are already "in flight"
/// (e.g. in an mpsc channel) will still be made available.
pub trait Close {
    /// Close it.
    fn close(&mut self);
}

impl<T> Close for mpsc::Receiver<T> {
    fn close(&mut self) {
        <mpsc::Receiver<T>>::close(self)
    }
}

impl<T> Close for mpsc::UnboundedReceiver<T> {
    fn close(&mut self) {
        <mpsc::UnboundedReceiver<T>>::close(self)
    }
}

impl<I, S: Stream + Close> Close for Infallible<I, S> {
    fn close(&mut self) {
        self.stream.close()
    }
}

/// A wrapper around a [`Stream`] which can be closed.
#[derive(Debug)]
pub(crate) struct Closable<S> {
    stream: S,
    killcord: oneshot::Receiver<()>,
    closed: bool
}

impl<S: Stream + Close> Closable<S> {
    /// Create a new closable stream and return it along with the
    /// [`KillCord`] to close the stream.
    pub(crate) fn new(s: S) -> (KillCord, Self) {
        let (tx, rx) = oneshot::channel();
        let stream = Closable { stream: s, killcord: rx, closed: false };
        (KillCord(tx), stream)
    }
}

impl<S: Stream + Close> Stream for Closable<S> {
    type Item = S::Item;
    type Error = S::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if !self.closed {
            match self.killcord.poll() {
                Ok(Async::NotReady) => {}
                Ok(Async::Ready(())) | Err(oneshot::Canceled) => {
                    self.stream.close();
                    self.closed = true
                }
            }
        }
        self.stream.poll()
    }
}

/// Item of the [`Stream`] impls returned by [`stoppable()`] and [`closable()`].
#[derive(Debug, Clone)]
pub enum Event<I, T, E> {
    /// Item of the stream.
    Item(I, T),
    /// The end of the stream.
    End(I),
    /// The stream produced an error.
    Error(I, E)
}

/// A wrapper around a [`Stream`] that lifts the stream items/errors to [`Event`]s.
#[derive(Debug)]
struct Infallible<I, S>{
    id: I,
    stream: S,
    done: bool // `true` iff the original stream returned `None`
}

impl<I: Clone, S: Stream> Infallible<I, S> {
    /// Create a new infallible stream from the given stream.
    fn new(id: I, stream: S) -> Self {
        Infallible { id, stream, done: false }
    }
}

impl<I: Clone, S: Stream> Stream for Infallible<I, S> {
    type Item = Event<I, S::Item, S::Error>;
    type Error = Void;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if self.done {
            return Ok(Async::Ready(None))
        }
        match self.stream.poll() {
            Err(e) => Ok(Async::Ready(Some(Event::Error(self.id.clone(), e)))),
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Ok(Async::Ready(None)) => {
                self.done = true;
                Ok(Async::Ready(Some(Event::End(self.id.clone()))))
            }
            Ok(Async::Ready(Some(x))) => Ok(Async::Ready(Some(Event::Item(self.id.clone(), x))))
        }
    }
}