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};

pub struct AmqpRawConsumer {
    queue: QueueSettings,
    prefetch: u16,
}

impl AmqpRawConsumer {
    pub fn new(queue: QueueSettings) -> Self {
        AmqpRawConsumer {
            queue,
            prefetch: 5,
        }
    }

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

    // TODO deduplicate
    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;
                }
            }
        }
    }
}
