holly 0.5.3

use crate::{error::Error, fold::{self, dynamic_fold}};
use futures::{
    future::{self, Either::{A, B}, Executor},
    prelude::*,
    sink,
    sync::mpsc
};
use parking_lot::Mutex;
use std::{fmt, marker::PhantomData, sync::{atomic::{AtomicUsize, Ordering}, Arc}};
use void::Void;

/// A type which responds to messages.
pub trait Actor<T, E> where Self: Sized {
    /// The result of handling a message, yielding a new processing state.
    type Result: Future<Item = State<Self, T>, Error = E>;

    /// Handles a message and returns the processing state the actor transitions to.
    fn process(self, ctx: &mut Context<T>, msg: Option<T>) -> Self::Result;
}

/// The processing state an actor transitions to after receiving a message.
pub enum State<A, T> {
    /// The actor is ready to accept more messages.
    Ready(A),

    /// The actor is ready to accept more messages and wants to add
    /// another [`Stream`] of messages to its mailbox.
    Stream(A, Box<dyn Stream<Item = T, Error = Void> + Send>),

    /// The actor wants to close its mailbox.
    ///
    /// This will prevent other actors from sending any more messages to
    /// this actor's address. All enqueued messages will still be
    /// processed. Note that it is the actor's responsibility to close
    /// streams added via [`State::Stream`].
    Close(A),

    /// The actor has finished processing (terminal state).
    Done
}

impl<A, T> fmt::Debug for State<A, T> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            State::Ready(..) => f.write_str("State::Ready"),
            State::Stream(..) => f.write_str("State::Stream"),
            State::Close(..) => f.write_str("State::Close"),
            State::Done => f.write_str("State::Done")
        }
    }
}

impl<A, T> Into<fold::State<A, T, ()>> for State<A, T> {
    fn into(self) -> fold::State<A, T, ()> {
        match self {
            State::Ready(a) => fold::State::Ready(a),
            State::Stream(a, s) => fold::State::Stream(a, s),
            State::Close(a) => fold::State::Close(a),
            State::Done => fold::State::Done(())
        }
    }
}

/// The ID of an actor.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct Id(usize);

impl Id {
    pub fn as_usize(self) -> usize {
        self.0
    }
}

impl fmt::Display for Id {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        self.0.fmt(f)
    }
}

/// The address of an actor.
///
/// Allows sending messages to the actor.
pub struct Addr<T, A = T> {
    id: Id,
    outgoing: mpsc::Sender<T>,
    marker: PhantomData<A>
}

impl<T, A: Into<T>> Addr<T, A> {
    /// Get address/actor ID.
    pub fn id(&self) -> Id {
        self.id
    }

    /// Asynchronously send a message to the actor of this address.
    ///
    /// Will fail if the recipient no longer exists.
    /// If the actor's mailbox is full, the future will wait until there is room.
    pub fn send(self, msg: A) -> impl Future<Item = Self, Error = Error> {
        SendOne(self.id, self.outgoing.send(msg.into()), self.marker)
    }

    /// Immediately send a message to the actor of this address.
    ///
    /// Used outside of futures.
    /// Fails if the actor's mailbox is full or the recipient no longer exists.
    pub fn send_now(&mut self, msg: A) -> Result<(), Error> {
        self.outgoing.try_send(msg.into()).map_err(|e| {
            if e.is_full() {
                return Error::MailboxFull
            }
            if e.is_disconnected() {
                return Error::ActorGone
            }
            unreachable!("Impossible TrySendError")
        })
    }

    /// Create an `Addr<T, B>` from this `Addr<T, A>`.
    ///
    /// This enables the creation of addresses which only accept a
    /// strict subset of messages the original address accepts.
    pub fn cast<B: Into<T>>(&self) -> Addr<T, B> {
        Addr {
            id: self.id,
            outgoing: self.outgoing.clone(),
            marker: PhantomData
        }
    }
}

impl<T, A> Clone for Addr<T, A> {
    fn clone(&self) -> Self {
        Addr {
            id: self.id,
            outgoing: self.outgoing.clone(),
            marker: self.marker
        }
    }
}

impl<T, A> fmt::Debug for Addr<T, A> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("Addr").field("id", &self.id).finish()
    }
}

/// Future delivering a message to the actor.
///
/// Actor mailboxes are bounded and if full, this future will wait until there is room.
#[derive(Debug)]
#[must_use = "futures do nothing unless polled"]
pub struct SendOne<T, A>(Id, sink::Send<mpsc::Sender<T>>, PhantomData<A>);

impl<T, A> Future for SendOne<T, A> {
    type Item = Addr<T, A>;
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match self.1.poll() {
            Ok(Async::Ready(s)) =>
                Ok(Async::Ready(Addr {
                    id: self.0,
                    outgoing: s,
                    marker: self.2
                })),
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Err(_) => Err(Error::ActorGone)
        }
    }
}

/// The execution context passed to an `Actor`.
///
/// It allows access to this actor's ID and to the scheduler to spawn other actors.
#[derive(Debug)]
pub struct Context<T> {
    /// Actor's own address.
    addr: Addr<T>,
    /// Global scheduler running actors.
    exec: Scheduler
}

impl<T> Context<T> {
    /// The actor ID.
    pub fn id(&self) -> Id {
        self.addr.id()
    }

    /// The address of the actor having access to this context.
    pub fn address(&self) -> Addr<T> {
        self.addr.clone()
    }

    /// The global scheduler to spawn new actors.
    pub fn scheduler(&self) -> &Scheduler {
        &self.exec
    }
}

/// A failure report.
#[derive(Debug)]
pub struct Fail<E> {
    /// Failure ID.
    id: Id,
    /// Error causing this failure.
    error: E
}

impl<E> Fail<E> {
    /// The failure ID corresponding to the [`Actor`] that caused the error.
    pub fn id(&self) -> Id {
        self.id
    }

    /// The actual error causing this failure.
    pub fn error(&self) -> &E {
        &self.error
    }

    /// Deconstruct into error.
    pub fn into_error(self) -> E {
        self.error
    }
}

/// Options for spawning an actor.
#[derive(Debug)]
pub struct Options<F> {
    /// Maximum mailbox size.
    backlog: u32,
    /// Supervisor, which is notified of errors.
    supervisor: Option<Addr<F>>
}

impl<F> Default for Options<F> {
    fn default() -> Self {
        Self { backlog: 1000, supervisor: None }
    }
}

impl<F> Options<F> {
    /// Set maximum mailbox size.
    pub fn backlog(self, n: u32) -> Self {
        Self { backlog: n, .. self }
    }

    /// Set supervisor which is notified of actor errors.
    pub fn supervisor(self, p: Addr<F>) -> Self {
        Self { supervisor: Some(p), .. self }
    }
}

/// Executor which processes all actors.
#[derive(Clone)]
pub struct Scheduler {
    exec: Arc<dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>,
    counter: Arc<AtomicUsize>
}

impl Scheduler {
    /// Create a new scheduler with the given [`Executor`].
    pub fn new<E>(e: E) -> Self
    where
        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync + 'static
    {
        Scheduler {
            exec: Arc::new(e),
            counter: Arc::new(AtomicUsize::new(0))
        }
    }

    /// Create a new scheduler with the given [`Executor`].
    ///
    /// Since the executor is not required to implement [`Sync`], access to it
    /// will be guarded by a `Mutex`.
    pub fn single_threaded<E>(e: E) -> Self
    where
        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + 'static
    {
        struct SyncExec<E>(Mutex<E>);

        impl<E, F> Executor<F> for SyncExec<E>
        where
            E: Executor<F>,
            F: Future<Item = (), Error = ()>
        {
            fn execute(&self, future: F) -> Result<(), future::ExecuteError<F>> {
                self.0.lock().execute(future)
            }
        }

        Scheduler::new(SyncExec(Mutex::new(e)))
    }

    /// Creates a new actor.
    ///
    /// The actor will be spawned onto the executor and messages will
    /// be delivered to it as long as it does not transition into the
    /// [`State::Done`] processing state.
    ///
    /// The spawned actor has no supervisor and its mailbox has a
    /// capacity of 1000 messages. Use [`Scheduler::spawn_ext`] to configure a
    /// different value or to set a supervisor.
    pub fn spawn<A, T, E>(&self, a: A) -> Result<Addr<T>, Error>
    where
        A: Actor<T, E> + Send + 'static,
        T: Send + 'static,
        E: Send + 'static,
        <A as Actor<T, E>>::Result: Send + 'static
    {
        let opts: Options<Fail<E>> = Default::default();
        self.spawn_ext(a, opts)
    }

    /// Creates a new actor with additional options.
    ///
    /// The actor will be spawned onto the executor and messages will
    /// be delivered to it as long as it does not transition into the
    /// [`State::Done`] processing state.
    pub fn spawn_ext<A, T, E, F>(&self, a: A, opts: Options<F>) -> Result<Addr<T>, Error>
    where
        A: Actor<T, E> + Send + 'static,
        T: Send + 'static,
        E: Send + 'static,
        F: From<Fail<E>> + Send + 'static,
        <A as Actor<T, E>>::Result: Send + 'static
    {
        let id = Id(self.counter.fetch_add(1, Ordering::Relaxed));

        let (tx_main, rx_main) = mpsc::channel(opts.backlog as usize);

        let addr = Addr {
            id,
            outgoing: tx_main,
            marker: PhantomData
        };

        let mut ctx = Context { addr: addr.clone(), exec: self.clone() };

        let future = dynamic_fold(a, rx_main, move |actor, item| {
                actor.process(&mut ctx, item).map(Into::into)
            })
            .map(|_| ())
            .or_else(move |e| {
                if let Some(p) = opts.supervisor {
                    let f = Fail { id, error: e };
                    A(p.send(f.into()).then(|_| Ok(())))
                } else {
                    B(future::ok(()))
                }
            });

        self.exec.execute(Box::new(future)).map_err(|_| Error::SpawnError)?;

        Ok(addr)
    }

    /// Execute some [`Future`].
    ///
    /// This allows actors to spawn auxiliary tasks onto this [`Scheduler`].
    pub fn execute<F>(&self, future: F) -> Result<(), Error>
    where
        F: Future<Item = (), Error = ()> + Send + 'static
    {
        self.exec.execute(Box::new(future)).map_err(|_| Error::SpawnError)
    }
}

impl fmt::Debug for Scheduler {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("Scheduler")
    }
}