Tag Archive APC

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

1. config file ก่อนตามระเบียบ
   ?

   \application\config\RabbitMQ.php

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57

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

2. ถึงคิวไฟล์ driver ที่จะเป็นตัวกลางระหว่าง codeigniter และ php-amqplib
   ?

   \application\libraries\RabbitMQ\RabbitMQ.php

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45

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

3. ที่นี้ก็ controller ที่จะทำงานให้
   ?

   \application\controllers\Messaging.php

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170

   <?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();       }   }

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