<?php

namespace App\Lib\Common;


use App\QueueLog;
use Illuminate\Http\Request;
use Illuminate\Support\Arr;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPProtocolChannelException;
use PhpAmqpLib\Exception\AMQPProtocolConnectionException;
use PhpAmqpLib\Exception\AMQPProtocolException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMQ {

    private $connection = null;
    private $channel = null;

    public function __construct()
    {
        if (false) { //false === $this->checkIfRabbitMQInstalled()
            if(false !== $this->createConnection()) {
                // connection created successfully
                $this->createChannel();
            }
        }
    }

    private function checkIfRabbitMQInstalled() {
        try {
            $output = shell_exec('whereis rabbitmq-server');
            if (!empty($output) && stripos($output,':') !== false) {
                $package = explode(':',$output);
                if (!empty($package[1])) {
                    return true;
                }
            }
            throw new \Exception("Rabbitmq server not installed");
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    private function createConnection() {
        try {
            if (is_null($this->connection)) {
                $this->connection = AMQPStreamConnection::create_connection([
                    ['host' => _env('AMQP_HOST'),
                    'port' => _env('AMQP_PORT'),
                    'user' => _env('AMQP_USER'),
                    'password' => _env('AMQP_PASS'),
                    'vhost' => _env('AMQP_VHOST')]
                ]);
            }
            return true;
        } catch(\InvalidArgumentException $e){
            report($e);
            return false;
        } catch(AMQPTimeoutException $e){
            report($e);
            return false;
        }catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    private function createChannel() {
        try {
            if (!is_null($this->connection)) {
                $this->channel = $this->connection->channel();

                $this->channel->set_nack_handler(
                    function(AMQPMessage $message)
                    {
                        //\Log::info(json_encode($message->body),['context' => 'message_nack', 'status' => 'failure']);
                        $body = json_decode($message->body, true);
                        //\Log::info($body);
                        if(!empty(Arr::get($body,'data.job_id',''))) {
                            QueueLog::where('job_id', Arr::get($body,'data.job_id',''))->update(['status' => 'pen_nack']);
                        }
                    }
                );
                $this->channel->set_ack_handler(
                    function(AMQPMessage $message)
                    {
                        //\Log::info(json_encode($message->body),['context' => 'message_ack', 'status' => 'success']);
                        $body = json_decode($message->body, true);
                        //\Log::info($body);
                        if(!empty(Arr::get($body,'data.job_id',''))) {
                            QueueLog::where('job_id', Arr::get($body,'data.job_id',''))->update(['status' => 'pen_ack']);
                        }

                    }
                );
                $this->channel->set_return_listener(
                    function ($replyCode, $replyText, $exchange, $routingKey, AMQPMessage $message) {
                        \Log::info(json_encode($message->body),['context' => 'message_return', 'status' => 'success']);
                    }
                );
                $this->channel->confirm_select();
                return true;
            }
            throw new \Exception(RabbitMQ::class." can not create a connection");
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    private function declareExchange($exchange, $exchange_type) {
        try {
            if(!is_null($this->channel)) {
                /*
                 * name: $exchange
                 * type: direct
                 * passive: false // don't check if an exchange with the same name exists
                 * durable: true // the exchange will survive server restarts
                 * auto_delete: false //the exchange won't be deleted once the channel is closed.
                 */
                $this->channel->exchange_declare($exchange, $exchange_type, false, true, false);
                return true;
            }
            throw new \Exception(RabbitMQ::class." No channel found while declaring exchange");
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    private function declareQueue($queue_name) {
        try {
            if(!is_null($this->channel)) {
                /*
                 * name: $queue
                 * passive: false
                 * durable: true // the queue will survive server restarts
                 * exclusive: false // the queue can be accessed in other channels
                 * auto_delete: false //the queue won't be deleted once the channel is closed.
                 */
                $this->channel->queue_declare($queue_name, false, true, false, false);
                return true;
            }
            throw new \Exception(self::class." No channel found while declaring queue");
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }


    private function bindQueue($queue, $exchange, $exchange_key) {
        try {
            if(!is_null($this->channel)) {
                /*
                 * name: $exchange
                 * type: direct
                 * passive: false // don't check if an exchange with the same name exists
                 * durable: true // the exchange will survive server restarts
                 * auto_delete: false //the exchange won't be deleted once the channel is closed.
                 */
                $this->channel->queue_bind($queue, $exchange, $exchange_key);
                return true;
            }
            throw new \Exception(RabbitMQ::class." No channel found while binding queue");
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    public function queueJob(array $data, $exchange, $exchange_key, $exchange_type, $queue = '', QueueLog $queueLog, $shutdown = false) {
        return false;
        if (is_null($this->channel)) {
            if(true !== $ret = $this->createChannel()) {
                //@todo: mark the job as failed reason connection can not established
                return $ret;
            }
        }

        if(true !== $exch = $this->declareExchange($exchange, $exchange_type)) {
            //@todo: mark the job as failed reason exchange can not declared
            return $exch;
        }

        if (!empty($queue)) {
            if(true !== $q = $this->bindQueue($queue, $exchange, $exchange_key)) {
                //@todo: mark the job as failed reason queue can not bind
                return $q;
            }
        }

        try {
            $msg = new AMQPMessage(json_encode($data), array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

            $this->channel->basic_publish($msg, $exchange, $exchange_key, true, false);

            $this->channel->wait_for_pending_acks();
            if(true === $shutdown) {
                $this->shutDown();
            }
            return true;
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    public function basicConsume($queue_name, $consumer_tag = '', $q_bind = false, $exchange = '', $route_key = '', $callback = null) {
        try {
            $this->declareQueue($queue_name);
            if (true === $q_bind) {
                if (empty($exchange) || empty($route_key)) {
                    return "Exchange name and route key parameter is required as queue bind is true";
                }
                $this->bindQueue($queue_name, $exchange, $route_key);
            }
            if(is_null($callback)) {
                $callback = self::class.'::callMeBack';
            }
            /*
             * queue: Queue from where to get the messages
             * consumer_tag: Consumer identifier
             * no_local: Don't receive messages published by this consumer.
             * no_ack: If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.
             * exclusive: Request exclusive consumer access, meaning only this consumer can access the queue
             * nowait:
             * callback: A PHP Callback
             */
            $this->channel->basic_consume($queue_name, $consumer_tag, false, false, false, false, $callback);
            return true;
        } catch (\Exception $e) {
            report($e);
            return false;
        }
    }

    static function callMeBack($msg) {
        try {
            //\Log::info($msg->body);
            $data = json_decode($msg->body, true);
            if (Arr::has($data,'data',false)) {
                $data = Arr::get($data,'data',[]);
            }
            $job_handler = '';
            if (Arr::has($data,'job_handler',false)) {
                $job_handler = Arr::get($data,'job_handler','');
                unset($data['job_handler']);
            } else if(Arr::has($data,'job_id',false)) {
                $queueLogObj = QueueLog::find(Arr::get($data,'job_id',''));
                if(!empty($queueLogObj->job_handler)) {
                    $job_handler = $queueLogObj->job_handler;
                }
                unset($queueLogObj);
            }
            if (!empty($job_handler)) {
                $handler = [];
                $handler = explode('@', $job_handler);
                $handler_class = $handler[0];
                $handler_function = $handler[1];
                //\Log::info("handler Class ".$handler_class. " method ".$handler_function);
                if(class_exists($handler_class)) {
                    if(method_exists($handler_class, $handler_function)) {
                        $output = call_user_func([$handler_class, $handler_function], $data);
                        //make sure $output is always in json format
                        QueueLog::markCompleted(Arr::get($data,'job_id',''), $output);
                        //$msg->delivery_info ['channel']->basic_ack($msg->delivery_info ['delivery_tag']);
                        return true;
                    }
                    throw new \Exception("Method ".$handler_function." not found in handler class ".$handler_class);
                }
                throw new \Exception("Handler class ".$handler_class." not found");
            }
            throw new \Exception("No job handler found");
        } catch (\Exception $e) {
            report($e);
            $msg->delivery_info ['channel']->basic_ack($msg->delivery_info ['delivery_tag']);
        } catch (AMQPProtocolChannelException $e) {
            report($e);
            $msg->delivery_info ['channel']->basic_ack($msg->delivery_info ['delivery_tag']);
        } catch (AMQPProtocolConnectionException $e) {
            report($e);
            $msg->delivery_info ['channel']->basic_ack($msg->delivery_info ['delivery_tag']);
        } catch (AMQPProtocolException $e) {
            report($e);
            $msg->delivery_info ['channel']->basic_ack($msg->delivery_info ['delivery_tag']);
        }
    }

    public function loopCallback() {
        if (is_null($this->channel)) {
            return "No channel found";
        }
        while($this->channel->callbacks) {
            $this->channel->wait();
        }
        return true;
    }

    public function getChannel() {
        if(is_null($this->channel)) {
            $this->createChannel();
        }
        return $this->channel;
    }

    public function closeChannel() {
        if(!is_null($this->channel)) {
            $this->channel->close();
        }
    }

    public function closeConnection() {
        if(!is_null($this->connection)) {
            $this->connection->close();
        }
    }

    public function shutDown() {
        $this->closeChannel();
        $this->closeConnection();
    }
}