use std::error;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

use crate::broker::keeper::Keeper;

/// Data struct combining the topic to send to and the protobuf received
#[derive(Debug)]
pub struct CollectedData {
    pub topic: String,
    pub data: Vec<u8>,
}

impl CollectedData {
    /// Returns a CollectedData instance with the topic specified
    ///
    /// # Arguments
    ///
    /// * `topic` - The topic to use
    ///
    pub fn new(topic: &str) -> Self {
        Self {
            topic: String::from(topic),
            data: Vec::new(),
        }
    }
}

/// A collector allows receiving and collecting messages from a redis broker, from the specified channels
///
pub struct Collector {
    sender: mpsc::Sender<CollectedData>,
    keeper: Arc<Mutex<Keeper>>,
    channels: Vec<String>,
}

impl Collector {
    /// Returns a Collector instance with the specified options
    ///
    /// # Arguments
    ///
    /// * `sender` - A sender for a CollectedData channel. The tx endpoint of the communication channel between the collector and
    ///    the publisher
    /// * `keeper`: A Keeper wrapped by a Mutex wrapped by an Arc. Allowing multiple threads to access the Keeper
    /// * `channels`: A set of channels to subscribe to
    ///
    fn new(sender: mpsc::Sender<CollectedData>, keeper: Arc<Mutex<Keeper>>, channels: Vec<String>) -> Self {
        Self { sender, keeper, channels }
    }

    /// Returns the result of the execution. During execution the collector will connect to the broker,
    /// subscribe to channels, continuously read messages from the broker and send them on the channel
    ///
    fn execute(&self) -> Result<(), Box<dyn error::Error>> {
        let guard = Arc::new(AtomicBool::new(true));
        match self.keeper.lock() {
            Ok(mut k) => k.add_guard(guard.clone()),
            Err(e) => eprintln!("unable to lock Keeper: {e:#?}"),
        }

        let client = redis::Client::open("redis://127.0.0.1/")?;
        let mut connection = client.get_connection()?;
        let mut pubsub = connection.as_pubsub();

        for channel in &self.channels {
            pubsub.subscribe(channel)?;
        }

        while guard.load(Ordering::SeqCst) {
            let msg = pubsub.get_message()?;
            let mut data = CollectedData::new(msg.get_channel_name());
            data.data.extend_from_slice(msg.get_payload_bytes());
            if let Err(e) = self.sender.send(data) {
                eprintln!("unable to send data over channel: {e:#?}");
            }
        }

        Ok(())
    }
}
/// Returns the thread JoinHandle for the thread created
///
/// # Arguments
///
/// * `sender`: A sender for a CollectedData channel. The tx endpoint of the communication channel between the collector module and
///    the publisher
/// * `keeper`: A Keeper wrapped by a Mutex wrapped by an Arc. Allowing multiple threads to access the Keeper
/// * `channels`: A set of channels to subscribe to
///
pub fn execute(
    sender: mpsc::Sender<CollectedData>,
    keeper: Arc<Mutex<Keeper>>, channels: Vec<String>,
) -> thread::JoinHandle<()> {
    thread::spawn(|| {
        let collector = Collector::new(sender, keeper, channels);

        if let Err(e) = collector.execute() {
            eprintln!("collector stopped execution: {e:#?}");
        }
    })
}
