php PhpAmqpLib-Connection-AMQPConnection类(方法)实例源码

/**
  * Process incoming request to generate pdf invoices and send them through 
  * email.
  */
 public function listen()
 {
     $this->log->addInfo('Begin listen routine');
     $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
     $channel = $connection->channel();
     $channel->queue_declare('invoice_queue', false, true, false, false);
     /**
      * don't dispatch a new message to a worker until it has processed and 
      * acknowledged the previous one. Instead, it will dispatch it to the 
      * next worker that is not still busy.
      */
     $channel->basic_qos(null, 1, null);
     /**
      * indicate interest in consuming messages from a particular queue. When they do 
      * so, we say that they register a consumer or, simply put, subscribe to a queue.
      * Each consumer (subscription) has an identifier called a consumer tag
      */
     $channel->basic_consume('invoice_queue', '', false, false, false, false, array($this, 'process'));
     $this->log->addInfo('Consuming from queue');
     while (count($channel->callbacks)) {
         $this->log->addInfo('Waiting for incoming messages');
         $channel->wait();
     }
     $channel->close();
     $connection->close();
 }

/**
  * getChannel
  *
  * @param string $connection
  *
  * @return AMQPChannel
  */
 protected function getChannel($connection)
 {
     if (isset($this->channels[$connection])) {
         return $this->channels[$connection];
     }
     if (!isset($this->connections[$connection])) {
         throw new \InvalidArgumentException(sprintf('Unknown connection "%s". Available: [%s]', $connection, implode(', ', array_keys($this->connections))));
     }
     if (!isset($this->channels[$connection])) {
         $this->channels[$connection] = array();
     }
     if (isset($this->connections[$connection]['ssl']) && $this->connections[$connection]['ssl']) {
         if (empty($this->connections[$connection]['ssl_options'])) {
             $ssl_opts = array('verify_peer' => true);
         } else {
             $ssl_opts = array();
             foreach ($this->connections[$connection]['ssl_options'] as $key => $value) {
                 if (!empty($value)) {
                     $ssl_opts[$key] = $value;
                 }
             }
         }
         $conn = new AMQPSSLConnection($this->connections[$connection]['host'], $this->connections[$connection]['port'], $this->connections[$connection]['login'], $this->connections[$connection]['password'], $this->connections[$connection]['vhost'], $ssl_opts);
     } else {
         $conn = new AMQPConnection($this->connections[$connection]['host'], $this->connections[$connection]['port'], $this->connections[$connection]['login'], $this->connections[$connection]['password'], $this->connections[$connection]['vhost']);
     }
     //$conn->connect();
     $this->channels[$connection] = $conn->channel();
     return $this->channels[$connection];
 }

protected function setupConnection()
 {
     Yii::trace('Connecting to broker...', __METHOD__);
     $this->connection = new AMQPConnection($this->host, $this->port, $this->user, $this->password, $this->vhost, $this->insist, $this->login_method, $this->login_response, $this->locale, $this->connection_timeout, $this->read_write_timeout, $this->context);
     $this->channel = $this->connection->channel();
     $this->channel->queue_declare($this->queue, false, true, false, false);
 }

function processvideo()
 {
     $this->load->config('amqp');
     $exchange = 'video';
     $queue = 'video_q';
     $consumer_tag = 'consumer';
     $connection = new AMQPConnection($this->config->item('host'), $this->config->item('port'), $channel = $this->config->item('user'), $channel = $this->config->item('pass'), "/");
     $channel = $connection->channel();
     $channel->queue_declare($queue, false, true, false, false);
     $callback = function ($msg) {
         print_r($msg);
         die;
         $collection = $this->mongo_db->db->selectCollection('video');
         $result = $collection->update(["video_id" => $msg->video_id], ["status" => "processing"]);
         sleep(range(5, 10));
         $start_date = new DateTime('2000-01-01 ' . $msg->start_time);
         $since_start = $start_date->diff(new DateTime('2012-09-11 ' . $msg->end_time));
         $video_len = $since_start->h . ":" . $since_start->i . ":" . $since_start->s;
         $result = $collection->update(["video_id" => $msg->video_id], ["status" => "done", "link" => "https://youtube.com?dffd", "video_len" => $video_len]);
         $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
     };
     $channel->basic_qos(null, 1, null);
     $channel->basic_consume($queue, '', false, false, false, false, $callback);
     while (count($channel->callbacks)) {
         $channel->wait();
     }
     $channel->close();
     $connection->close();
 }

protected function getConnection()
 {
     $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
     $channel = $connection->channel();
     $channel->queue_declare('email_queue', false, false, false, false);
     return $channel;
 }

/**
  * Create service with name
  *
  * @param ServiceLocatorInterface $serviceLocator
  * @param $name
  * @param $requestedName
  * @return mixed
  */
 public function createServiceWithName(ServiceLocatorInterface $serviceLocator, $name, $requestedName)
 {
     $name = substr($requestedName, strlen(self::MANAGER_PREFIX));
     $config = $serviceLocator->get('config');
     $workerConfig = $config['workers'][$name];
     if (!$serviceLocator->has($workerConfig['manager']['director'])) {
         throw new \RuntimeException("Could not load {$workerConfig['manager']['director']}");
     }
     $director = $serviceLocator->get($workerConfig['manager']['director']);
     if (!$director instanceof DirectorInterface) {
         throw new \RuntimeException("Could not load {$workerConfig['manager']['director']}");
     }
     $rabbitMqSettings = isset($workerConfig['rabbitmq_settings']) ? $workerConfig['rabbitmq_settings'] : $config['rabbitmq_settings'];
     $conn = new AMQPConnection($rabbitMqSettings['host'], $rabbitMqSettings['port'], $rabbitMqSettings['username'], $rabbitMqSettings['password']);
     // Bind to the generic exchange
     $eventChannel = $conn->channel();
     $exchange = $workerConfig['manager']['general']['exchange']['name'];
     $exchangeType = $workerConfig['manager']['general']['exchange']['type'];
     $eventQueueName = $workerConfig['manager']['general']['queue']['name'];
     $routingKey = isset($workerConfig['manager']['general']['queue']['routing_key']) ? $workerConfig['manager']['general']['queue']['routing_key'] : null;
     $eventChannel->queue_declare($eventQueueName, false, true, false, false);
     $eventChannel->exchange_declare($exchange, $exchangeType, false, false, true);
     $eventChannel->queue_bind($eventQueueName, $exchange, $routingKey);
     // Bind to the one specific for the workers
     $processorQueueName = $workerConfig['manager']['worker']['queue']['name'];
     $workerExchange = $workerConfig['manager']['worker']['exchange']['name'];
     $workerExchangeType = $workerConfig['manager']['worker']['exchange']['type'];
     $workerChannel = $conn->channel();
     $workerChannel->exchange_declare($workerExchange, $workerExchangeType, false, false, false);
     $workerChannel->queue_declare($processorQueueName, false, true, false, false);
     $workerChannel->queue_bind($processorQueueName, $workerExchange);
     return new DirectedManager($conn, $eventChannel, $eventQueueName, $workerChannel, $workerExchange, $processorQueueName, $director);
 }

public function notasAction()
 {
     $this->mqconf = new \Phalcon\Config\Adapter\Ini(CONFIG_PATH . DIRS . "ini/mq.ini");
     $connection = new AMQPConnection($this->mqconf->host, $this->mqconf->port, $this->mqconf->user, $this->mqconf->pasw);
     $channel = $connection->channel();
     $channel->queue_declare(SITESLUG . '_eliminar_noticia', false, true, false, false);
     echo ' [*] Servicio listener para despublicacion ,  CTRL+C para cancelar', "\n";
     $callback = function ($msg) {
         try {
             $node = json_decode($msg->body);
             var_dump($node);
             $publicacion = new \Rpp\Services\Unpublish\Noticia((int) $node->nid);
             $di = new Phalcon\DI();
             $di->set('viewCache', $this->viewCache);
             $publicacion->setDI($di);
             $publicacion->builder();
         } catch (\Exception $e) {
             var_dump($e);
         }
         break;
     };
     $channel->basic_consume(SITESLUG . '_eliminar_noticia', '', false, true, false, false, $callback);
     while (count($channel->callbacks)) {
         $channel->wait();
     }
 }

public function publish_msg($data)
 {
     $conn = new AMQPConnection(HOST, PORT, USER, PASS, VHOST);
     $ch = $conn->channel();
     $msg = new AMQPMessage(json_encode($data), array('content_type' => 'text/plain', 'delivery_mode' => 2));
     $rtn = $ch->basic_publish($msg, "inStock");
 }

public function makeAction()
 {
     $this->mqconf = new \Phalcon\Config\Adapter\Ini(CONFIG_PATH . DIRS . "ini/mq.ini");
     $connection = new AMQPConnection($this->mqconf->host, $this->mqconf->port, $this->mqconf->user, $this->mqconf->pasw);
     $channel = $connection->channel();
     $channel->queue_declare(SITESLUG . '_sondeo', false, true, false, false);
     echo ' [*] Servicio listener de sondeos iniciado ,  CTRL+C para cancelar', "\n";
     $callback = function ($msg) {
         try {
             $vars = json_decode($msg->body);
             var_dump($vars);
             \Rpp\Services\Get\Elecciones::$pattern = null;
             \Rpp\Services\Get\Elecciones::$sondeos = null;
             $sondeo = \Rpp\Services\Get\Elecciones::get($vars->id);
             var_dump($sondeo);
         } catch (\Exception $e) {
             var_dump($e);
         }
         break;
     };
     $channel->basic_consume(SITESLUG . '_sondeo', '', false, true, false, false, $callback);
     while (count($channel->callbacks)) {
         $channel->wait();
     }
 }

public function portadaAction()
 {
     $this->mqconf = new \Phalcon\Config\Adapter\Ini(CONFIG_PATH . DIRS . "ini/mq.ini");
     $connection = new AMQPConnection($this->mqconf->host, $this->mqconf->port, $this->mqconf->user, $this->mqconf->pasw);
     $channel = $connection->channel();
     $channel->queue_declare(SITESLUG . '_publicacion_destacados', false, true, false, false);
     echo ' [*] Servicio listener para la publicacion de destacados ,  CTRL+C para cancelar', "\n";
     $callback = function ($msg) {
         try {
             $node = json_decode($msg->body);
             print_r($node);
             $destacados = new \Rpp\Services\Publish\Destacados(@$node->slug);
             $di = new Phalcon\DI();
             $di->set('viewCache', $this->viewCache);
             $destacados->setDI($di);
             $destacados->load();
         } catch (\Exception $e) {
             var_dump($e);
         }
         break;
     };
     $channel->basic_consume(SITESLUG . '_publicacion_destacados', '', false, true, false, false, $callback);
     while (count($channel->callbacks)) {
         $channel->wait();
     }
 }

public function procesoAction()
 {
     $this->mqconf = new \Phalcon\Config\Adapter\Ini(CONFIG_PATH . DIRS . "ini/mq.ini");
     $connection = new AMQPConnection($this->mqconf->host, $this->mqconf->port, $this->mqconf->user, $this->mqconf->pasw);
     $channel = $connection->channel();
     $channel->queue_declare(SITESLUG . '_widget', false, true, false, false);
     echo ' [*] Servicio listener para widgets ,  CTRL+C para cancelar', "\n";
     $callback = function ($msg) {
         try {
             $data = json_decode($msg->body);
             echo $data->widget;
             $method = "widget_{$data->widget}";
             unset($data->widget);
             if (method_exists($this, $method)) {
                 $this->{$method}($data);
             }
         } catch (\Exception $e) {
             var_dump($e);
         }
     };
     $channel->basic_consume(SITESLUG . '_widget', '', false, true, false, false, $callback);
     while (count($channel->callbacks)) {
         $channel->wait();
     }
 }

/**
  * Perform the check
  *
  * @see \ZendDiagnostics\Check\CheckInterface::check()
  * @return Failure|Success
  */
 public function check()
 {
     if (!class_exists('PhpAmqpLib\\Connection\\AMQPConnection')) {
         return new Failure('PhpAmqpLib is not installed');
     }
     $conn = new AMQPConnection($this->host, $this->port, $this->user, $this->password, $this->vhost);
     $conn->channel();
     return new Success();
 }

function publish($data)
{
    $conn = new AMQPConnection(HOST, PORT, USER, PASS, VHOST);
    $ch = $conn->channel();
    $ch->queue_declare("oversold_queue", false, true, false, false);
    $ch->queue_bind("oversold_queue", "oversold");
    $msg = new AMQPMessage(json_encode($data), array('content_type' => 'text/plain', 'delivery_mode' => 2));
    $rtn = $ch->basic_publish($msg, "oversold");
}

/**
  * Set up AMQP connection.
  */
 public function setUp()
 {
     $container = $this->getContainer();
     $exchangeName = 'general';
     $connection = new AMQPConnection($container->getParameter('ongr_task_messenger.publisher.default.amqp.host'), $container->getParameter('ongr_task_messenger.publisher.default.amqp.port'), $container->getParameter('ongr_task_messenger.publisher.default.amqp.user'), $container->getParameter('ongr_task_messenger.publisher.default.amqp.password'));
     $this->channel = $connection->channel();
     list($queueName, , ) = $this->channel->queue_declare();
     $this->channel->queue_bind($queueName, $exchangeName, explode('.', gethostname())[0]);
     $this->channel->basic_consume($queueName, getmypid(), false, true, true, true, [$this, 'verifyMessage']);
 }

/**
  * Sends an invoice generation task to the workers
  * 
  * @param int $invoiceNum
  */
 public function execute($invoiceNum)
 {
     $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
     $channel = $connection->channel();
     $channel->queue_declare('invoice_queue', false, true, false, false);
     $msg = new AMQPMessage($invoiceNum, array('delivery_mode' => 2));
     $channel->basic_publish($msg, '', 'invoice_queue');
     $channel->close();
     $connection->close();
 }

public static function execute($entry)
 {
     if (!empty($entry)) {
         $connection = new AMQPConnection('impact.ccat.eu', 5672, 'myjar', 'myjar');
         $channel = $connection->channel();
         $msg = new AMQPMessage($entry, array('content_type' => 'application/json'));
         $channel->basic_publish($msg, '', 'solved-interest-queue');
         $channel->close();
         $connection->close();
     }
 }

function send($nachricht)
 {
     $connection = new AMQPConnection('141.22.29.97', 5672, 'invoiceSender', 'invoiceSender');
     $channel = $connection->channel();
     $channel->queue_declare('controllerInvoice', false, false, false, false);
     settype($nachricht, "string");
     $msg = new AMQPMessage($nachricht);
     $channel->basic_publish($msg, '', 'controllerInvoice');
     $channel->close();
     $connection->close();
 }

/**
  * Sends an invoice generation task to the workers
  * 
  * @param int $invoiceNum
  */
 public function execute($invoiceNum)
 {
     $this->log->addInfo('Received invoice for processing: ' . $invoiceNum);
     $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
     $channel = $connection->channel();
     $channel->queue_declare('invoice_queue', false, true, false, false);
     $msg = new AMQPMessage($invoiceNum, array('delivery_mode' => 2));
     $channel->basic_publish($msg, '', 'invoice_queue');
     $this->log->addInfo('Published task to worker');
     $channel->close();
     $connection->close();
 }

public function main()
 {
     $exchange = 'default';
     $queue = 'default';
     $consumer_tag = 'consumer';
     $conn = new AMQPConnection(RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASS, RABBITMQ_VHOST);
     $ch = $conn->channel();
     $ch->queue_declare($queue, false, true, false, false);
     $ch->exchange_declare($exchange, 'direct', false, true, false);
     $ch->queue_bind($queue, $exchange);
     function process_message($msg)
     {
         $msg_body = json_decode($msg->body, true);
         if (isset($msg_body['params']) && is_array($msg_body['params'])) {
             $msg_body['params'] = json_encode($msg_body['params']);
         }
         $msg_body = array_values($msg_body);
         $dispatcher = new ShellDispatcher($msg_body, false);
         try {
             $dispatcher->dispatch();
         } catch (Exception $e) {
             RabbitMQ::publish(json_decode($msg->body, true), ['exchange' => 'requeueable', 'queue' => 'requeueable_messages']);
             $newMessage[] = $msg->body;
             $newMessage[] = '==>';
             $newMessage[] = $e->getMessage();
             $newMessage[] = $e->getFile();
             $newMessage[] = $e->getLine();
             $newMessage[] = $e->getTraceAsString();
             $newMessage[] = $e->getCode();
             $newMessage[] = $e->getPrevious();
             RabbitMQ::publish($newMessage, ['exchange' => 'unprocessed', 'queue' => 'unprocessed_messages']);
             EmailSender::sendEmail('elisio.leonardo@gmail.com', $msg->body, $newMessage);
         }
         $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
         // Send a message with the string "quit" to cancel the consumer.
         if ($msg->body === 'quit') {
             $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
         }
     }
     $ch->basic_qos(null, 1, null);
     $ch->basic_consume($queue, $consumer_tag, false, false, false, false, 'process_message');
     function shutdown($ch, $conn)
     {
         $ch->close();
         $conn->close();
     }
     register_shutdown_function('shutdown', $ch, $conn);
     // Loop as long as the channel has callbacks registered
     while (count($ch->callbacks)) {
         $ch->wait();
     }
 }

public function actionIndex($name, $message)
 {
     // Create a connection with RabbitMQ server.
     $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
     $channel = $connection->channel();
     // Create and publish the message to the exchange.
     $data = array('type' => 'chat', 'data' => array('name' => $name, 'message' => $message, 'dateTime' => date('d/m/Y H:i:s')));
     $message = new AMQPMessage(json_encode($data));
     $channel->basic_publish($message, 'chats');
     // Close connection.
     $channel->close();
     $connection->close();
 }