use async_std::prelude::StreamExt;
use icee_container_rs::get;
use lapin::Consumer;
use lapin::options::{
    BasicAckOptions, BasicConsumeOptions, BasicNackOptions, BasicQosOptions, QueueDeclareOptions,
};
use lapin::types::FieldTable;
use log::{debug, error};

use crate::{Message, QueueSettings, rabbitmq::RabbitMq};

/// Consumer callback processing each message
pub type DeliveryCallback = dyn Fn(Message) -> Result<(), String>;

/// AmqpConsumer holds open channel and consumes all messages from given queue
///
/// Does not support to consume from multiple queues,
/// if required use two Consumers with the same callback
///
/// ```no_run
/// # #[async_std::main]
/// # async fn main() {
/// use icee_rabbit_rs::{QueueSettings, AmqpConsumer};
///
/// let queue = QueueSettings::new("queue".into());
/// let callback = |_| { Ok(()) };
///
/// let consumer = AmqpConsumer::new(queue, Box::new(callback));
/// consumer.consume().await;
/// # }
/// ```
pub struct AmqpConsumer {
    callback: Box<DeliveryCallback>,
    queue: QueueSettings,
    prefetch: u16,
}

impl AmqpConsumer {
    pub fn new(queue: QueueSettings, callback: Box<DeliveryCallback>) -> Self {
        AmqpConsumer {
            callback,
            queue,
            prefetch: 5,
        }
    }

    pub fn with_prefetch(mut self, prefetch: u16) -> Self {
        self.prefetch = prefetch;
        self
    }

    /// Starts the consumer sending each message into specified callback
    ///
    /// In case of disconnect, consumer awaits for reconnect to continue without yielding an error
    /// or panicking
    pub async fn consume(&self) {
        loop {
            let mut consumer = self.connect().await;
            debug!("Starting consumer for queue [name={}]", self.queue.name);

            while let Some(delivery) = consumer.next().await {
                if let Ok((_channel, delivery)) = delivery {
                    match (self.callback)(Message::from(&delivery)) {
                        Ok(_) => {
                            if let Err(err) = delivery.ack(BasicAckOptions::default()).await {
                                error!("Ack message: {}", err.to_string());

                                break;
                            }
                        }
                        Err(e) => {
                            error!("Consumer process: {}", e);
                            let mut opt = BasicNackOptions::default();
                            opt.requeue = true;
                            let _res = delivery.nack(opt).await;
                        }
                    }
                }
            }
        }
    }

    pub async fn connect(&self) -> Consumer {
        loop {
            let conn = get::<RabbitMq>("rabbitmq")
                .await
                .expect("missing rabbitmq service");

            let guard = conn.connection().read().await;
            if let Ok(channel) = guard.create_channel().await {
                channel
                    .queue_declare(
                        &self.queue.name,
                        QueueDeclareOptions::from(&self.queue),
                        FieldTable::default(),
                    )
                    .await
                    .expect("declare channel");

                channel
                    .basic_qos(self.prefetch, BasicQosOptions::default())
                    .await
                    .expect("prefetch configuration");

                if let Ok(consumer) = channel
                    .basic_consume(
                        &self.queue.name,
                        "",
                        BasicConsumeOptions::default(),
                        FieldTable::default(),
                    )
                    .await
                {
                    return consumer;
                }
            }
        }
    }
}
