/**

 * 消息列队服务

 * @author zhou.tingze

 * @example

 * -----------------------------------Create----------------------------------------

 * $array = array('a','b','c','d');

 * $this->load->library('amqp_service');

 * $this->amqp_service->setSaveType('test_exchange', 'test_queue', 'test_router');

 * $this->amqp_service->createMessageQueue($array);

 * -----------------------------------End-------------------------------------------

 *

 * -----------------------------------Get-------------------------------------------

 * $this->load->library('amqp_service');

 * $this->amqp_service->setSaveType('test_exchange', 'test_queue', 'test_router');

 * $message_queue = $this->amqp_service->getMessageQueue();

 * var_dump($message_queue)

 * -----------------------------------End-------------------------------------------

 */

class Amqp_service extends Base_service{

	public $conn;

	public $exchange;

	public $queue;

	public $router;

    function __construct(){

        parent:: __construct();

        //获取系统配置

        $this->load->config('app_config', TRUE);

        $app_config = $this->config->item('app_config');

        $this->connect($app_config['amqp']);

    }

    /**

     *

     * 尝试连接Amqp服务

     */

    private function connect($amqp_args)

    {

    	$this->conn = new AMQPConnection($amqp_args);

		$this->conn->connect();

		if (!$this->conn->isConnected())

		{

			throw new Exception('Cannot connect to the broker.');

		}

    }

    /**

     *

     * 设定消息列队保存方式

     * @param String $exchange_name 交换机名

     * @param String $queue_name    消息列队名

     * @param String $router_name   路由名

     */

    public function setSaveType($exchange_name, $queue_name, $router_name)

    {

    	$this->exchange = $exchange_name;

    	$this->queue    = $queue_name;

    	$this->router   = $router_name;

    }

    /**

     *

     * 创建消息列队

     * @param Array $array

     */

    public function createMessageQueue($array)

    {

    	//创建交换机

		$channel = new AMQPChannel($this->conn);

		$ex      = new AMQPExchange($channel);

		//交换机名

		$ex->setName($this->exchange);

		$ex->setType(AMQP_EX_TYPE_DIRECT);

		$ex->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);

		$ex->declare();

		//创建消息列队

		$q = new AMQPQueue($channel);

		//队列名

		$q->setName($this->queue);

		$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);

		$q->declare();

		//绑定交换机与队列，并指定路由键

		$q->bind($this->exchange, $this->router);

		//消息发布

		$channel->startTransaction();

		$message = json_encode($array);

		$ex->publish($message, $this->router);

		$channel->commitTransaction();

		//$this->conn->disconnect();

    }

    /**

     *

     * 获取消息列队

     */

    public function getMessageQueue()

    {

    	try

    	{

	    	//设置queue名称，使用exchange，绑定routingkey

			$channel = new AMQPChannel($this->conn);

			$q       = new AMQPQueue($channel);

			$q->setName($this->queue);

			$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);

			$q->declare();

			$q->bind($this->exchange, $this->router);   

			//消息获取

			$messages = $q->get(AMQP_AUTOACK) ;

			$arr = array();

			if ($messages){

				$arr = json_decode($messages->getBody(), true );

			}

    	}catch (Exception $e){

    		throw new Exception($e->getMessage());

    	}

		//$this->conn->disconnect();

		return $arr;

    }

    /*

    public function getAllMessageQueue()

    {

		//设置queue名称，使用exchange，绑定routingkey

		$channel = new AMQPChannel($this->conn);

		$q       = new AMQPQueue($channel);

		$q->setName($this->queue);

		$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);

		$q->declare();

		$q->bind($this->exchange, $this->router);

		$this->conn->disconnect();

		//阻塞模式获取消息列队

		while(True){

		    $q->consume('processMessage');

		    //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答

		}

    }

    */

    public function __destruct()

    {

    	$this->conn->disconnect();

    }

}

/**

 * 消费回调函数

 * 处理消息

 * @param Object $envelope

 * @param Object $queue

 */

/*

function processMessage($envelope, $queue) {

    $msg = $envelope->getBody();

    echo $msg . '<br />';

    //手动发送ACK应答

    $queue->ack($envelope->getDeliveryTag());

}

*/