php PhpAmqpLib-Channel-AMQPChannel类(方法)实例源码

/**
  * Gets publisher AMQ channel
  *
  * @return AMQPChannel
  */
 protected function getChannel()
 {
     if (null === $this->channel) {
         $this->channel = $this->amq->channel();
         $this->channel->exchange_declare($this->exchange, 'topic', false, true, false);
     }
     return $this->channel;
 }

function it_releases_job_onto_rabbitmq(AMQPChannel $channel, RabbitMQQueue $queue)
 {
     // delete
     $channel->basic_ack('fooTagId')->shouldBeCalled();
     // release with attempts added into body
     $queue->later(1, 'foo', [0 => "someData", "attempts" => 1], 'default')->shouldBeCalled();
     $this->release(1);
 }

/**
  * Creates (if not yet created) and returns an AMQP channel.
  *
  * @return \PhpAmqpLib\Channel\AMQPChannel
  */
 protected function getChannel()
 {
     if (null === $this->channel) {
         $this->channel = $this->connection->channel();
         $this->channel->queue_declare($this->queueName, false, false, false, false);
         $this->channel->basic_qos(null, 1, null);
     }
     return $this->channel;
 }

/**
  * @return AMQPChannel
  */
 public function getAmqpChannel()
 {
     if ($this->amqp_channel) {
         return $this->amqp_channel;
     }
     $this->amqp_channel = $this->connection->getAmqpConnection()->channel();
     $this->amqp_channel->queue_declare($this->queue_config['queue_name'], false, $is_durable = true, false, false);
     $this->amqp_channel->basic_qos(null, $this->queue_config['fetch_count'], null);
     return $this->amqp_channel;
 }

/**
  * Initializes the message queue class
  *
  * @param \PhpAmqpLib\Channel\AMQPChannel $channel AMQP channel
  * @param string $queue Message queue name
  * @throws \Aimeos\MW\MQueue\Exception
  */
 public function __construct(\PhpAmqpLib\Channel\AMQPChannel $channel, $queue)
 {
     try {
         $channel->queue_declare($queue, false, true, false, false);
         $channel->basic_qos(null, 1, null);
     } catch (\Exception $e) {
         throw new \Aimeos\MW\MQueue\Exception($e->getMessage());
     }
     $this->channel = $channel;
     $this->queue = $queue;
 }

public function __construct($name, AMQPChannel $channel, $options)
 {
     $this->name = $name;
     $this->channel = $channel;
     $this->type = isset($options["type"]) ? $options["type"] : "direct";
     $this->passive = isset($options["passive"]) ? $options["passive"] : false;
     $this->durable = isset($options["durable"]) ? $options["durable"] : true;
     $this->auto_delete = isset($options["auto_delete"]) ? $options["auto_delete"] : false;
     if ($name != "") {
         $channel->exchange_declare($name, $this->type, $this->passive, $this->durable, $this->auto_delete);
     }
 }

public function testDestruct()
 {
     $this->channelMock->expects($this->once())->method('close');
     $this->clientMock->expects($this->once())->method('close');
     $queue = new MessageQueue($this->clientMock);
     unset($queue);
 }

/**
  * @param DomainMessage $domainMessage
  */
 private function publishWithAMQP(DomainMessage $domainMessage)
 {
     $payload = $domainMessage->getPayload();
     $eventClass = get_class($payload);
     $this->logger->info("publishing message with event type {$eventClass} to exchange {$this->exchange}");
     $this->channel->basic_publish($this->messageFactory->createAMQPMessage($domainMessage), $this->exchange);
 }

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

/**
  * Close the running connection
  */
 public function close()
 {
     if (null !== $this->connection && $this->connection->isConnected()) {
         $this->channel->close();
         $this->connection->close();
     }
 }

/**
  * Test declaring a queue on the channel using the configured queue name
  */
 public function testQueueDeclaration()
 {
     $qName = 'queue_name';
     $this->setStubConfigData(['getQueueName' => $qName, 'getQueueConfiguration' => ['queue' => $qName, 'passive' => false, 'durable' => true, 'exclusive' => true, 'auto_delete' => true, 'nowait' => true]]);
     $this->channel->expects($this->once())->method('queue_declare')->with($this->identicalTo($qName));
     $this->invokeRestrictedMethod($this->amqpApi, 'declareQueue');
 }

protected function setUpAmqp()
 {
     $this->conn = new AMQPSocketConnection($_SERVER['AMQP_HOST'], $_SERVER['AMQP_PORT'], $_SERVER['AMQP_USER'], $_SERVER['AMQP_PASS'], $_SERVER['AMQP_VHOST']);
     $this->channel = $this->conn->channel();
     $this->channel->exchange_declare('event_band.test.exchange', 'topic');
     $this->channel->queue_declare('event_band.test.event');
     $this->channel->queue_bind('event_band.test.event', 'event_band.test.exchange', 'event.#');
 }

/**
  * Consumes one event and calls callback for it.
  *
  * @param integer $timeout Optional timeout in seconds. Default is no timeout.
  *
  * @return void
  */
 public function consume($timeout = 0)
 {
     try {
         $this->channel->wait(null, false, $timeout);
     } catch (AMQPTimeoutException $e) {
         return;
     }
 }

private function getChannel()
 {
     if ($this->channel) {
         return $this->channel;
     }
     $this->channel = $this->connection->channel();
     $this->channel->queue_declare($this->queue, false, false, false, false);
     return $this->channel;
 }

/**
  * Destructor.
  *
  * Closes RabbitMQ connection and channel if necessary.
  */
 public function __destruct()
 {
     if ($this->channel) {
         $this->channel->close();
     }
     if ($this->connection) {
         $this->connection->close();
     }
 }

public function __destroy()
 {
     if ($this->_channel) {
         $this->_channel->close();
     }
     if ($this->getConnection && $this->getConnection->isConnected()) {
         $this->getConnection->close();
     }
 }

public function tearDown()
 {
     if ($this->ch2) {
         $this->ch2->close();
     }
     if ($this->conn) {
         $this->conn->close();
     }
 }

/**
  * @param $exchange_name
  */
 private function connect($exchange_name)
 {
     if (null !== $this->channel) {
         return;
     }
     $this->channel = $this->connection->channel();
     $this->channel->exchange_declare($exchange_name, 'fanout', false, true, false);
     $this->channel->queue_declare($exchange_name, false, true, false, false);
     $this->channel->queue_bind($exchange_name, $exchange_name);
 }

public function tearDown()
 {
     if ($this->ch) {
         $this->ch->exchange_delete($this->exchange_name);
         $this->ch->close();
     }
     if ($this->conn) {
         $this->conn->close();
     }
 }

/**
  *
  * {@inheritDoc}
  *
  */
 protected function write(array $record)
 {
     $data = $record["formatted"];
     $routingKey = sprintf('%s.%s', substr($record['level_name'], 0, 4), $record['channel']);
     if ($this->exchange instanceof AMQPExchange) {
         $this->exchange->publish($data, strtolower($routingKey), 0, array('delivery_mode' => 2, 'Content-type' => 'application/json'));
     } else {
         $this->exchange->basic_publish(new AMQPMessage((string) $data, array('delivery_mode' => 2, 'content_type' => 'application/json')), $this->exchangeName, strtolower($routingKey));
     }
 }