<?php if ( ! defined('BASEPATH')) exit('No direct script access allowed');
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

class Amqp
{
	private $connection;
	private $host;
	private $port;
	private $user;
	private $pass;
	private $vhost;
	private $channel;
	private $message;
	private $delay;
	private $delayed_queue_name;
	private $delayed_exchange_name;
	private $delayed_base_time;

	public function __construct() {
		$ci = &get_instance();

		$ci->config->load('amqp', TRUE);
        $amqp = $ci->config->item('amqp');

        $this->host 	= $amqp['amqp_host'];
        $this->port 	= $amqp['amqp_port'];
        $this->user 	= $amqp['amqp_user'];
        $this->pass 	= $amqp['amqp_pass'];
        $this->vhost 	= $amqp['amqp_vhost'];

        $this->queue 	= 'default_queue';
        $this->exchange = 'default_exchange';
        $this->dead_letter = 'delayed';
        $this->is_delayed = false;
        $this->init_called = false;

        $this->delayed_queue_name = $amqp['amqp_delayed_queue_name'];
		$this->delayed_exchange_name = $amqp['amqp_delayed_exchange_name'];
		$this->delayed_base_time = $amqp['amqp_delayed_base_time'];
	}

	public function set_vhost() {

	}

	public function delayedQueueName()
	{
		return $this->delayed_queue_name;
	}

	public function delayedExchangeName()
	{
		return $this->delayed_exchange_name;
	}

	public function delayedBaseTime()
	{
		return $this->delayed_base_time;
	}

	public function init_direct() {
		$this->init_called = true;
	}

	public function init_delayed() {
		$this->init_called = true;
		$this->is_delayed = true;
	}

	public function init_connection() {
		$this->connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->pass, '/'.$this->vhost);
		return $this;
	}

	public function create_channel() {
		$this->channel = $this->connection->channel();
		return $this;
	}

	public function queue($queue) {
		$this->queue = ($this->is_delayed === true) ? $this->dead_letter.'_'.$queue : $queue;
	}

	public function exchange($exchange) {
		$this->exchange = ($this->is_delayed === true) ? $this->dead_letter.'_'.$exchange : $exchange;
	}

	public function delay($ms) {
		$this->delayed = $ms; // in milisecond
	}

	public function set_delay()
	{
		$headers = new AMQPTable(array('x-delay' => $this->delayed));
		$this->message_header = $headers;
	}

	public function set_queue($queue_name, $passive = false, $durable = false, $exclusive = false, $auto_delete = false) {
		$this->channel->queue_declare($queue_name, $passive, $durable, $exclusive, $auto_delete);
		return $this;
	}

	public function set_exchange($exchange_name, $type = AMQPExchangeType::DIRECT, $passive = false, $durable = false, $auto_delete = false) {

		$this->channel->exchange_declare($exchange_name , $type, $passive, $durable, $auto_delete);
		return $this;
	}

	public function set_delayed_queue($queue_name, $passive = false, $durable = false, $exclusive = false, $auto_delete = false, $nowait = false) {
		$this->channel->queue_declare($queue_name, $passive, $durable, $exclusive, $auto_delete, $nowait, new AMQPTable(array('x-dead-letter-exchange' => $this->dead_letter)));
		return $this;
	}

	public function set_delayed_exchange($exchange_name, $type = 'x-delayed-message', $passive = false, $durable = true, $auto_delete = false, $internal = false, $nowait = false) {

		$this->channel->exchange_declare($exchange_name , $type, $passive, $durable, $auto_delete, $internal, $nowait, new AMQPTable(array('x-delayed-type' => AMQPExchangeType::FANOUT)));
		return $this;
	}

	public function bind() {

		$this->channel->queue_bind($this->queue, $this->exchange);
		return $this;
	}

	public function data($data)
	{
		$this->message = $data;
	}

	public function set_data($data) {

		if(is_array($data)) {
			$data 						= json_encode($data);
			$config['content_type'] 	= 'application/json';
		}else{
			$data 						= (string)$data;
			$config['content_type'] 	= 'text/plain';
		}
		$config['delivery_mode'] 	= ($this->is_delayed === true) ?  2 : AMQPMessage::DELIVERY_MODE_PERSISTENT;

		$msg = new AMQPMessage($data, $config);

		if(isset($this->message_header)) {
			$msg->set('application_headers', $this->message_header);
		}

		$this->message = $msg;
	}

	public function basic_publish() {
		$this->channel->basic_publish($this->message, $this->exchange);
		return $this;
	}

	public function exit() {
		$this->channel->close();
		$this->connection->close();
		return $this;
	}

	public function build() {
		if($this->init_called === false) {
			throw new Exception("Error Processing Request", 1);
			exit;
		}
		// $this->init_core();
		$this->init_connection();
		$this->create_channel();

		if($this->is_delayed === true){
			$this->set_delayed_queue($this->queue);
		}else{
			$this->set_queue($this->queue);
		}

		if($this->is_delayed === true){
			$this->set_delayed_exchange($this->exchange);
		}else{
			$this->set_exchange($this->exchange);
		}
		$this->bind($this->queue, $this->exchange);

		if($this->is_delayed === true){
			$this->set_delay();
		}

		$this->set_data($this->message);
		return $this;
	}

	public function send() {
		$this->build();
		$this->basic_publish();
		$this->exit();
	}


	public function contoh()
	{
		$queue 		= 'message';
        $exchange  	= 'router';

		$connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->pass, '/'.$this->vhost);  //, VHOST
		$channel = $connection->channel();

		$channel->queue_declare($queue, false, false, false, false);
		$channel->exchange_declare($exchange , AMQPExchangeType::DIRECT, false, false, false);
		$channel->queue_bind($queue, $exchange );

		$data = ['order_id' => 1, 'product_id' => 1, 'quantity' => 10, 'data_type' => 'processCheckout'];
		// $data = array_slice($data, 1);
        $data = json_encode($data);
        // echo $data;
        // exit;
        

		// $messageBody = implode(' ', $data);

		$msg = new AMQPMessage((string)$data);

		// $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
		$channel->basic_publish($msg, $exchange);
		echo 'Message sent';
		$channel->close();
		$connection->close();
	}
}