mod component;
mod builder;
mod message_sniffer;

pub use crate::message::*;
pub use self::component::{Component, ComponentLink};
pub use self::builder::SystemBuilder;
pub use self::message_sniffer::MessageSniffer;

use std::collections::HashMap;
use std::sync::*;
use std::thread;


#[cfg(test)]
mod programmable_message_producer;

#[cfg(test)]
use self::programmable_message_producer::*;

#[cfg(test)]
mod inspectable_message_consumer;

#[cfg(test)]
use self::inspectable_message_consumer::*;

struct RunningComponent {
    name: String,
    halt: Box<dyn Fn() -> ()>,
    join: Box<dyn FnOnce()>
}

/// The system runs components to process events of type T.
/// The system run all event producers and accumulate all events,
/// which are fed to event consumers. Events generated by consumers
/// are processed at the next run.
pub struct System {
    receiver: mpsc::Receiver<Message>,
    components: Vec<Box<dyn Component + Send>>
}

impl System {
    /// Create a new system which receives messages using 
    /// the given receiver.
    pub fn new(receiver: mpsc::Receiver<Message>) -> Self {
        Self {
            receiver,
            components: Vec::new()
        }
    }

    pub fn register_component(&mut self, component: Box<dyn Component + Send>) {
        self.components.push(component);
    }

    /// Create a map of message types and interested component's sender.
    /// This allow to statically manage an incoming message by pushing it to all senders
    /// for that specific message type.
    fn map_components(&self) -> HashMap<MessageType, Vec<mpsc::Sender<Message>>> {
        let mut map = HashMap::<MessageType, Vec<mpsc::Sender<Message>>>::new();

        for component in self.components.iter() {
            for message_type in component.listen_to() {
                if let Some(listeners) = map.get_mut(&message_type) {
                    listeners.push(component.get_sender());
                } else {
                    map.insert(message_type, vec!(component.get_sender()));
                }
            }
        }

        map
    }

    /// Starts all registered components, listen for an halt message
    /// then join all threads.
    pub fn run(&mut self) {
        let mut running_components = Vec::new();

        if self.components.is_empty() {
            log::warn!("No registered component");
            return;
        }

        let map = self.map_components();

        // start each component in its own thread
        for mut component in self.components.drain(..) {
            let name = component.get_name().to_string();
            let sender = component.get_sender();
            let handler = thread::spawn(move || component.run());

            log::info!("Starting component {}", &name);

            running_components.push(RunningComponent {
                name,
                halt: Box::new(move || { let _ = sender.send(Message::Halt); } ),
                join: Box::new(|| { handler.join().unwrap(); })
            });
        }
        
        loop {
            // Read incoming messages
            let data = self.receiver.recv();
            log::debug!("System received message {:?}", &data);

            if let Ok(message) = data {
                // Stop processing messages if a Halt message has been received
                if Message::Halt == message {
                    Self::broadcast_halt_message(&running_components);
                    break;
                }
                
                let k = MessageType::from(&message);

                // Send message to interested components
                if let Some(senders) = map.get(&k) {
                    log::debug!("System sending message {:?}", &message);
                    
                    for sender in senders {
                        sender.send(message.clone()).unwrap();
                    }
                } else {
                    log::debug!("No sender found for key {:?}", k);
                }
            }
        }

        Self::join_threads(running_components);
    }
    
    /// Send the terminate message to all components in order to
    /// stop them before shutting down.
    fn broadcast_halt_message(running_components: &[RunningComponent]) {
        for component in running_components.iter() {
            (component.halt)();
        }
    }

    /// Wait all threads to finish.
    fn join_threads(running_components: Vec<RunningComponent>) {
        log::info!("{}", "System waiting all threads to join");

        for component in running_components {
            (component.join)();
            log::info!("Component {} has joined system", component.name);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc::channel;
    use crate::system::*;
     
    #[test]
    fn when_no_component_is_registered_then_the_system_exits() {
        // Arrange

        let (_, bus_rx) = channel::<Message>();
        let mut sys = System::new(bus_rx);
        
        // Act

        sys.run();

        // Assert

        assert!(true);
    }

    #[test]
    fn when_system_receive_an_halt_message_then_exit() {

        // Arrange

        let halt_message: Message = Message::Halt;
        let messages = vec!(halt_message);
        
        let (bus_tx, bus_rx) = channel::<Message>();
        let mut sys = System::new(bus_rx);
        
        let pro = ProgrammableMessageProducer::new(bus_tx.clone(), messages);
        sys.register_component(Box::new(pro));

        // Act

        sys.run();

        // Assert

        // TODO: commented assert to be fixed?!
        assert!(true);
        // three messages have ben sent through the system
        // assert_eq!(3, cons.inspect().len());
    }

    #[test]
    fn messages_are_dispatched_to_registered_consumers() {

        // Arrange

        let error_message = Message::Error;
        let key_pressed_message = Message::KeyPressed(Key::GeneralReset);

        let messages = vec!(error_message.clone(), key_pressed_message.clone());
        let (bus_tx, bus_rx) = channel::<Message>();
        let mut sys = System::new(bus_rx);
        
        let pro = ProgrammableMessageProducer::new(bus_tx.clone(), messages);
        sys.register_component(Box::new(pro));

        let received = Arc::new(RwLock::new(Vec::new()));
        let cons = InspectableMessageConsumer::new(bus_tx.clone(), 
            vec!(MessageType::KeyPressed),
            received.clone());
        sys.register_component(Box::new(cons));

        // Act

        sys.run();

        // Assert

        // Consumer have received two messages

        let messages = received.read().unwrap();

        assert!(!messages.contains(&error_message));
        assert!(messages.contains(&key_pressed_message));
    }
}

