CodeIgniter: php-amqplib

ลองเขียน custom driver สำหรับติดต่อกับ RabbitMQ / AMQP ยังไม่สมบูรณ์แต่ก็ดีกว่าการที่ต้องมา connect ผ่าน php-amqplib ทุกครั้งที่จะใช้ตามปกติ

1. config file ก่อนตามระเบียบ[code language=”php” title=”\application\config\RabbitMQ.php”]<?php
   defined(‘BASEPATH’) or exit(‘No direct script access allowed’);

   /*
   | ——————————————————————-
   | RABBITMQ CONNECTIVITY SETTINGS
   | ——————————————————————-
   | This file will contain the settings needed to access your RabbitMQ.
   |
   | For complete instructions please consult the ‘RabbitMQ Connection’
   | page of the User Guide.
   |
   | ——————————————————————-
   | EXPLANATION OF VARIABLES
   | ——————————————————————-
   |
   | [‘connection_timeout’] => 3.0,
   | [‘context’] => null,
   | [‘heartbeat’] => 0
   | [‘host’],
   | [‘insist’] => false,
   | [‘keepalive’] => false,
   | [‘locale’] => ‘en_US’,
   | [‘login_method’] => ‘AMQPLAIN’,
   | [‘login_response’] => null,
   | [‘password’],
   | [‘port’],
   | [‘read_write_timeout’] => 3.0,
   | [‘user’],
   | [‘vhost’] => ‘/’,
   */
   $config[‘RabbitMQ’][‘connects’] = [
   ‘default’ => [
   ‘connection_timeout’ => 3.0,
   ‘context’ => null,
   ‘heartbeat’ => 0,
   ‘host’ => ‘127.0.0.1’,
   ‘insist’ => false,
   ‘keepalive’ => false,
   ‘locale’ => ‘en_US’,
   ‘login_method’ => ‘AMQPLAIN’,
   ‘login_response’ => null,
   ‘password’ => ‘guest’,
   ‘port’ => 5672,
   ‘read_write_timeout’ => 3.0,
   ‘user’ => ‘guest’,
   ‘vhost’ => ‘/’,
   ],
   ];

   /*
   | ——————————————————————-
   | RABBITMQ DEBUG SETTINGS
   | ——————————————————————-
   |
   */
   $config[‘RabbitMQ’][‘debug’] = false;
   [/code]

2. ถึงคิวไฟล์ driver ที่จะเป็นตัวกลางระหว่าง codeigniter และ php-amqplib[code language=”php” title=”\application\libraries\RabbitMQ\RabbitMQ.php”]<?php
   use PhpAmqpLib\Connection\AMQPStreamConnection;
   use PhpAmqpLib\Message\AMQPMessage;

   class RabbitMQ extends AMQPStreamConnection
   {

   public function __construct($config = array())
   {
   $this->CI = &get_instance();

   $this->config = $config[‘RabbitMQ’];

   if ($this->config[‘debug’] == true) {
   define(‘AMQP_DEBUG’, true);
   }
   }

   public function connect($dsnId = ‘default’) {
   $dsn = $this->config[‘connects’][$dsnId];

   $this->connection = new AMQPStreamConnection($dsn[‘host’], $dsn[‘port’], $dsn[‘user’], $dsn[‘password’], $dsn[‘vhost’], $dsn[‘insist’], $dsn[‘login_method’], $dsn[‘login_response’], $dsn[‘locale’], $dsn[‘connection_timeout’], $dsn[‘read_write_timeout’], $dsn[‘context’], $dsn[‘keepalive’], $dsn[‘heartbeat’]);

   $this->channel = $this->connection->channel();
   }

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

   public function setMessageJson($datas)
   {
   $msg_body = json_encode($datas);

   $properties = [
   ‘content_type’ => ‘application/json’,
   ‘delivery_mode’ => AMQPMessage::DELIVERY_MODE_PERSISTENT,
   ];

   return new AMQPMessage($msg_body, $properties);
   }

   }[/code]

3. ที่นี้ก็ controller ที่จะทำงานให้[code language=”php” title=”\application\controllers\Messaging.php”]<?php
   defined(‘BASEPATH’) or exit(‘No direct script access allowed’);

   class Messaging extends CI_Controller
   {
   public function __construct()
   {
   parent::__construct();

   $this->load->driver(‘RabbitMQ’);
   }

   public function receive()
   {
   set_time_limit(0);

   $this->rabbitmq->connect();

   $exchange_name = ‘customers’;
   $queue_name = ‘invoices’;

   /**
   * Declares exchange
   *
   * @param string $exchange_name
   * @param string $type
   * @param bool $passive
   * @param bool $durable
   * @param bool $auto_delete
   * @param bool $internal
   * @param bool $nowait
   * @param array $arguments
   * @param int $ticket
   * @return mixed|null
   */
   $this->rabbitmq->channel->exchange_declare($exchange_name, ‘fanout’, false, true, false);

   /**
   * Declares queue, creates if needed
   *
   * @param string $queue
   * @param bool $passive
   * @param bool $durable
   * @param bool $exclusive
   * @param bool $auto_delete
   * @param bool $nowait
   * @param array $arguments
   * @param int $ticket
   * @return mixed|null
   */
   list($queueName, $message_count, $consumer_count) = $this->rabbitmq->channel->queue_declare(”, false, false, true, false);
   $this->rabbitmq->channel->queue_bind($queue_name, $exchange_name);

   $callback = function ($msg) {

   $datas = json_decode($msg->body, true);
   fwrite(fopen(‘RabbitMQReceive.txt’, ‘a+’), print_r($datas, true));

   sleep(substr_count($msg->body, ‘.’));

   /* delete message */
   $msg->delivery_info[‘channel’]->basic_ack($msg->delivery_info[‘delivery_tag’]);
   };

   /**
   * Starts a queue consumer
   *
   * @param string $queue_name
   * @param string $consumer_tag
   * @param bool $no_local
   * @param bool $no_ack
   * @param bool $exclusive
   * @param bool $nowait
   * @param callback|null $callback
   * @param int|null $ticket
   * @param array $arguments
   * @return mixed|string
   */
   $this->rabbitmq->channel->basic_consume($queue_name, ”, false, false, false, false, $callback);

   while (count($this->rabbitmq->channel->callbacks)) {
   /**
   * Wait for some expected AMQP methods and dispatch to them.
   * Unexpected methods are queued up for later calls to this PHP
   * method.
   *
   * @param array $allowed_methods
   * @param bool $non_blocking
   * @param int $timeout
   * @throws \PhpAmqpLib\Exception\AMQPOutOfBoundsException
   * @throws \PhpAmqpLib\Exception\AMQPRuntimeException
   * @return mixed
   */
   $this->rabbitmq->channel->wait();
   }

   $this->rabbitmq->channel->close();
   $connection->close();
   }

   public function send()
   {
   $this->rabbitmq->connect();

   $exchange_name = ‘customers’;
   $queue_name = ‘invoices’;

   /**
   * Declares exchange
   *
   * @param string $exchange_name
   * @param string $type
   * @param bool $passive
   * @param bool $durable
   * @param bool $auto_delete
   * @param bool $internal
   * @param bool $nowait
   * @param array $arguments
   * @param int $ticket
   * @return mixed|null
   */
   $this->rabbitmq->channel->exchange_declare($exchange_name, ‘fanout’, false, true, false);

   /**
   * Declares queue, creates if needed
   *
   * @param string $queue
   * @param bool $passive
   * @param bool $durable
   * @param bool $exclusive
   * @param bool $auto_delete
   * @param bool $nowait
   * @param array $arguments
   * @param int $ticket
   * @return mixed|null
   */
   list($queueName, $message_count, $consumer_count) = $this->rabbitmq->channel->queue_declare($queue_name, false, true, false, false);

   $datas = [
   ‘rand’ => rand(0, 100),
   ‘time’ => date(‘Y-m-d H:m:s’),
   ];

   /**
   * Declares message
   *
   * @param array $datas
   * @return string
   */
   $msg = $this->rabbitmq->setMessageJson($datas);

   /**
   * Publishes a message
   *
   * @param AMQPMessage $msg
   * @param string $exchange
   * @param string $routing_key
   * @param bool $mandatory
   * @param bool $immediate
   * @param int $ticket
   */
   $this->rabbitmq->channel->basic_publish($msg, $exchange_name, $queue_name);

   echo ‘<br><pre>’ . print_r($datas, true), ‘</pre>’;

   $this->rabbitmq->disconnect();

   }

   }
   [/code]

4. ทดสอบโดยเรียก http://localhost/CodeIgniter-3.1.3/messaging/send และ http://localhost/CodeIgniter-3.1.3/messaging/receive