[amqp-lib] The context should allow to get the lib's channel.

fixes #146

Author: makasim

pkg/amqp-lib/AmqpContext.php

@@ -120,18 +120,18 @@ public function createConsumer(PsrDestination $destination)
             $queue = $this->createTemporaryQueue();
             $this->bind(new AmqpBind($destination, $queue, $queue->getQueueName()));
 
-            return new AmqpConsumer($this->getChannel(), $queue, $this->buffer, $this->config['receive_method']);
+            return new AmqpConsumer($this->getLibChannel(), $queue, $this->buffer, $this->config['receive_method']);
         }
 
-        return new AmqpConsumer($this->getChannel(), $destination, $this->buffer, $this->config['receive_method']);
+        return new AmqpConsumer($this->getLibChannel(), $destination, $this->buffer, $this->config['receive_method']);
     }
 
     /**
      * @return AmqpProducer
      */
     public function createProducer()
     {
-        $producer = new AmqpProducer($this->getChannel(), $this);
+        $producer = new AmqpProducer($this->getLibChannel(), $this);
         $producer->setDelayStrategy($this->delayStrategy);
 
         return $producer;
@@ -142,7 +142,7 @@ public function createProducer()
      */
     public function createTemporaryQueue()
     {
-        list($name) = $this->getChannel()->queue_declare('', false, false, true, false);
+        list($name) = $this->getLibChannel()->queue_declare('', false, false, true, false);
 
         $queue = $this->createQueue($name);
         $queue->addFlag(InteropAmqpQueue::FLAG_EXCLUSIVE);
@@ -155,7 +155,7 @@ public function createTemporaryQueue()
      */
     public function declareTopic(InteropAmqpTopic $topic)
     {
-        $this->getChannel()->exchange_declare(
+        $this->getLibChannel()->exchange_declare(
             $topic->getTopicName(),
             $topic->getType(),
             (bool) ($topic->getFlags() & InteropAmqpTopic::FLAG_PASSIVE),
@@ -172,7 +172,7 @@ public function declareTopic(InteropAmqpTopic $topic)
      */
     public function deleteTopic(InteropAmqpTopic $topic)
     {
-        $this->getChannel()->exchange_delete(
+        $this->getLibChannel()->exchange_delete(
             $topic->getTopicName(),
             (bool) ($topic->getFlags() & InteropAmqpTopic::FLAG_IFUNUSED),
             (bool) ($topic->getFlags() & InteropAmqpTopic::FLAG_NOWAIT)
@@ -184,7 +184,7 @@ public function deleteTopic(InteropAmqpTopic $topic)
      */
     public function declareQueue(InteropAmqpQueue $queue)
     {
-        list(, $messageCount) = $this->getChannel()->queue_declare(
+        list(, $messageCount) = $this->getLibChannel()->queue_declare(
             $queue->getQueueName(),
             (bool) ($queue->getFlags() & InteropAmqpQueue::FLAG_PASSIVE),
             (bool) ($queue->getFlags() & InteropAmqpQueue::FLAG_DURABLE),
@@ -202,7 +202,7 @@ public function declareQueue(InteropAmqpQueue $queue)
      */
     public function deleteQueue(InteropAmqpQueue $queue)
     {
-        $this->getChannel()->queue_delete(
+        $this->getLibChannel()->queue_delete(
             $queue->getQueueName(),
             (bool) ($queue->getFlags() & InteropAmqpQueue::FLAG_IFUNUSED),
             (bool) ($queue->getFlags() & InteropAmqpQueue::FLAG_IFEMPTY),
@@ -215,7 +215,7 @@ public function deleteQueue(InteropAmqpQueue $queue)
      */
     public function purgeQueue(InteropAmqpQueue $queue)
     {
-        $this->getChannel()->queue_purge(
+        $this->getLibChannel()->queue_purge(
             $queue->getQueueName(),
             (bool) ($queue->getFlags() & InteropAmqpQueue::FLAG_NOWAIT)
         );
@@ -232,7 +232,7 @@ public function bind(InteropAmqpBind $bind)
 
         // bind exchange to exchange
         if ($bind->getSource() instanceof InteropAmqpTopic && $bind->getTarget() instanceof InteropAmqpTopic) {
-            $this->getChannel()->exchange_bind(
+            $this->getLibChannel()->exchange_bind(
                 $bind->getTarget()->getTopicName(),
                 $bind->getSource()->getTopicName(),
                 $bind->getRoutingKey(),
@@ -241,7 +241,7 @@ public function bind(InteropAmqpBind $bind)
             );
             // bind queue to exchange
         } elseif ($bind->getSource() instanceof InteropAmqpQueue) {
-            $this->getChannel()->queue_bind(
+            $this->getLibChannel()->queue_bind(
                 $bind->getSource()->getQueueName(),
                 $bind->getTarget()->getTopicName(),
                 $bind->getRoutingKey(),
@@ -250,7 +250,7 @@ public function bind(InteropAmqpBind $bind)
             );
             // bind exchange to queue
         } else {
-            $this->getChannel()->queue_bind(
+            $this->getLibChannel()->queue_bind(
                 $bind->getTarget()->getQueueName(),
                 $bind->getSource()->getTopicName(),
                 $bind->getRoutingKey(),
@@ -271,7 +271,7 @@ public function unbind(InteropAmqpBind $bind)
 
         // bind exchange to exchange
         if ($bind->getSource() instanceof InteropAmqpTopic && $bind->getTarget() instanceof InteropAmqpTopic) {
-            $this->getChannel()->exchange_unbind(
+            $this->getLibChannel()->exchange_unbind(
                 $bind->getTarget()->getTopicName(),
                 $bind->getSource()->getTopicName(),
                 $bind->getRoutingKey(),
@@ -280,15 +280,15 @@ public function unbind(InteropAmqpBind $bind)
             );
             // bind queue to exchange
         } elseif ($bind->getSource() instanceof InteropAmqpQueue) {
-            $this->getChannel()->queue_unbind(
+            $this->getLibChannel()->queue_unbind(
                 $bind->getSource()->getQueueName(),
                 $bind->getTarget()->getTopicName(),
                 $bind->getRoutingKey(),
                 $bind->getArguments()
             );
             // bind exchange to queue
         } else {
-            $this->getChannel()->queue_unbind(
+            $this->getLibChannel()->queue_unbind(
                 $bind->getTarget()->getQueueName(),
                 $bind->getSource()->getTopicName(),
                 $bind->getRoutingKey(),
@@ -309,7 +309,7 @@ public function close()
      */
     public function setQos($prefetchSize, $prefetchCount, $global)
     {
-        $this->getChannel()->basic_qos($prefetchSize, $prefetchCount, $global);
+        $this->getLibChannel()->basic_qos($prefetchSize, $prefetchCount, $global);
     }
 
     /**
@@ -336,7 +336,7 @@ public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
             }
         };
 
-        $consumerTag = $this->getChannel()->basic_consume(
+        $consumerTag = $this->getLibChannel()->basic_consume(
             $consumer->getQueue()->getQueueName(),
             $consumer->getConsumerTag(),
             (bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOLOCAL),
@@ -366,10 +366,10 @@ public function unsubscribe(InteropAmqpConsumer $consumer)
 
         $consumerTag = $consumer->getConsumerTag();
 
-        $this->getChannel()->basic_cancel($consumerTag);
+        $this->getLibChannel()->basic_cancel($consumerTag);
 
         $consumer->setConsumerTag(null);
-        unset($this->subscribers[$consumerTag], $this->getChannel()->callbacks[$consumerTag]);
+        unset($this->subscribers[$consumerTag], $this->getLibChannel()->callbacks[$consumerTag]);
     }
 
     /**
@@ -407,7 +407,7 @@ public function consume($timeout = 0)
     /**
      * @return AMQPChannel
      */
-    private function getChannel()
+    public function getLibChannel()
     {
         if (null === $this->channel) {
             $this->channel = $this->connection->channel();