* handle the change of СonsumerTag implemented

Author: ssserj

doc/ChangeLog

@@ -1,3 +1,10 @@
+2011-11-18	Sergey S. Sergeev
+
+	* main/Utils/AMQP/AMQPConsumer.class.php,
+	main/Utils/AMQP/AMQPDefaultConsumer.class.php,
+	main/Utils/AMQP/Pecl/AMQPPeclChannel.class.php:
+	handle change СonsumerTag implemented.
+
 2011-11-18	Evgeny V. Kokovikhin
 
 	* main/Messages/TextMessage.class.php, main/Messages/TextFileReceiver.class.php:

main/Utils/AMQP/AMQPConsumer.class.php

@@ -41,6 +41,15 @@ public function handleConsumeOk($consumerTag);
 		**/
 		public function handleCancelOk($consumerTag);
 
+		/**
+		 * Called when the consumer is changed tag
+		 *
+		 * @param string $fromTag
+		 * @param string $toTag
+		 * @return void
+		**/
+		public function handleChangeConsumerTag($fromTag, $toTag);
+
 		/**
 		 * @return AMQPConsumer
 		**/

main/Utils/AMQP/AMQPDefaultConsumer.class.php

@@ -72,6 +72,14 @@ public function handleDelivery(AMQPIncomingMessage $delivery)
 			// no work to do
 		}
 
+		/**
+		 * @return void
+		**/
+		public function handleChangeConsumerTag($fromTag, $toTag)
+		{
+			// no work to do
+		}
+
 		/**
 		 * @return AMQPDefaultConsumer
 		**/

main/Utils/AMQP/Pecl/AMQPPeclChannel.class.php

@@ -485,11 +485,12 @@ public function getNextDelivery()
 			}
 
 			$this->checkCommandResult(
-				is_array($messages),
+				is_array($messages) && !empty($messages),
 				"Could not consume from queue"
 			);
 
 			$message = array_shift($messages);
+
 			$incoming = AMQPIncomingMessage::spawn($message);
 
 			if ($this->consumer->getConsumerTag() === null) {
@@ -499,11 +500,9 @@ public function getNextDelivery()
 				$this->consumer->getConsumerTag()
 				!= $incoming->getConsumerTag()
 			) {
-				throw new WrongStateException(
-					"Consumer change tag consumerTag="
-					."{$this->consumer->getConsumerTag()}, "
-					."message.consumerTag={$incoming->getConsumerTag()}, "
-					."message.body={$incoming->getBody()}"
+				$this->consumer->handleChangeConsumerTag(
+					$this->consumer->getConsumerTag(),
+					$incoming->getConsumerTag()
 				);
 			}
 

test/main/Utils/AMQP/AMQPPeclTest.class.php

@@ -32,6 +32,11 @@ public function getCheckString()
 		{
 			return $this->checkString;
 		}
+
+		public function handleChangeConsumerTag($fromTag, $toTag)
+		{
+			return;
+		}
 	}
 
 	class AMQPPeclTest extends TestCase