use crate::{event::Event, transforms::Transform};
use futures01::{stream, Future, Stream as FutureStream};
use std::time::Duration;
use tokio01::timer::Interval;

/// A structure representing user-defined timer.
#[derive(Clone, Copy, Debug)]
pub struct Timer {
    pub id: u32,
    pub interval_seconds: u64,
}

/// A trait representing a runtime running user-defined code.
pub trait RuntimeTransform {
    /// Call user-defined "init" hook.
    fn hook_init<F>(&mut self, _emit_fn: F)
    where
        F: FnMut(Event) -> (),
    {
    }

    /// Call user-defined "process" hook.
    fn hook_process<F>(&mut self, event: Event, emit_fn: F)
    where
        F: FnMut(Event) -> ();

    /// Call user-defined "shutdown" hook.
    fn hook_shutdown<F>(&mut self, _emit_fn: F)
    where
        F: FnMut(Event) -> (),
    {
    }

    /// Call user-defined timer handler.
    fn timer_handler<F>(&mut self, _timer: Timer, _emit_fn: F)
    where
        F: FnMut(Event) -> (),
    {
    }

    /// Return (static) list of user-defined timers.
    fn timers(&self) -> Vec<Timer> {
        Vec::new()
    }
}

#[derive(Debug)]
enum Message {
    Init,
    Process(Event),
    Shutdown,
    Timer(Timer),
}

impl<T> Transform for T
where
    T: RuntimeTransform + Send,
{
    // used only in config tests (cannot be put behind `#[cfg(test)`])
    fn transform(&mut self, event: Event) -> Option<Event> {
        let mut out = Vec::new();
        self.transform_into(&mut out, event);
        assert!(out.len() <= 1);
        out.into_iter().next()
    }

    // used only in config tests (cannot be put behind `#[cfg(test)]`)
    fn transform_into(&mut self, output: &mut Vec<Event>, event: Event) {
        self.hook_process(event, |event| output.push(event));
    }

    fn transform_stream(
        mut self: Box<Self>,
        input_rx: Box<dyn FutureStream<Item = Event, Error = ()> + Send>,
    ) -> Box<dyn FutureStream<Item = Event, Error = ()> + Send>
    where
        Self: 'static,
    {
        let timers = self.timers();
        let input_rx: MessageStream = Box::new(
            input_rx
                .map(|event| Message::Process(event))
                .into_future()
                .map(move |(first, rest)| {
                    // The first message is always `Message::Init`.
                    let init_msg = stream::once(Ok(Message::Init));
                    // After it comes the first event, if any.
                    let first_event = first.map_or_else(
                        || -> MessageStream { Box::new(stream::empty()) },
                        |msg| -> MessageStream { Box::new(stream::once(Ok(msg))) },
                    );
                    // Then all other events followed by `Message::Shutdown` message
                    let rest_events_and_shutdown_msg =
                        rest.chain(stream::once(Ok(Message::Shutdown)));
                    // A stream of `Message::Timer(..)` events generated by timers.
                    let timer_msgs = make_timer_msgs_stream(timers);

                    init_msg
                        .chain(first_event)
                        .chain(rest_events_and_shutdown_msg.select(timer_msgs))
                })
                .map_err(|_| ())
                .into_stream()
                .flatten(),
        );

        let mut is_shutdown: bool = false; // TODO: consider using an enum describing the state instead of a
                                           // a single boolean variable.
                                           // It is used to prevent timers to emit messages after the source
                                           // stream stopped.
        Box::new(
            input_rx
                .map(move |msg| -> Vec<Event> {
                    let mut acc = Vec::new(); // TODO: create a stream adaptor to avoid buffering all events
                    if is_shutdown {
                        return acc;
                    }
                    match msg {
                        Message::Init => self.hook_init(|event| acc.push(event)),
                        Message::Process(event) => {
                            self.hook_process(event, |event| acc.push(event))
                        }
                        Message::Shutdown => {
                            self.hook_shutdown(|event| acc.push(event));
                            is_shutdown = true;
                        }
                        Message::Timer(timer) => self.timer_handler(timer, |event| acc.push(event)),
                    }
                    acc
                })
                .map(|v| stream::iter_ok(v))
                .flatten(),
        )
    }
}

type MessageStream = Box<dyn FutureStream<Item = Message, Error = ()> + Send>;

fn make_timer_msgs_stream(timers: Vec<Timer>) -> MessageStream {
    let mut stream: MessageStream = Box::new(stream::empty());
    for timer in timers {
        let interval_stream = Interval::new_interval(Duration::new(timer.interval_seconds, 0))
            .map(move |_| Message::Timer(timer))
            .map_err(|_| ());

        stream = Box::new(stream.select(interval_stream));
    }
    stream
}
