fix amqp orpahned message exception.

Author: makasim

src/SubscriptionConsumerConsumeUntilUnsubscribedSpec.php

@@ -6,6 +6,7 @@
 use Interop\Queue\PsrContext;
 use Interop\Queue\PsrMessage;
 use Interop\Queue\PsrQueue;
+use Interop\Queue\PsrSubscriptionConsumer;
 use Interop\Queue\PsrSubscriptionConsumerAwareContext;
 use PHPUnit\Framework\TestCase;
 
@@ -19,6 +20,11 @@ abstract class SubscriptionConsumerConsumeUntilUnsubscribedSpec extends TestCase
      */
     private $context;
 
+    /**
+     * @var PsrSubscriptionConsumer
+     */
+    protected $subscriptionConsumer;
+
     protected function tearDown()
     {
         if ($this->context) {
@@ -50,19 +56,20 @@ public function test()
             return true;
         };
 
-        $subscriptionConsumer = $context->createSubscriptionConsumer();
+        $this->subscriptionConsumer = $subscriptionConsumer = $context->createSubscriptionConsumer();
         $subscriptionConsumer->subscribe($fooConsumer, $callback);
         $subscriptionConsumer->subscribe($barConsumer, $callback);
 
         $subscriptionConsumer->consume(1000);
 
         $this->assertEquals(2, $consumedMessages);
 
+        $subscriptionConsumer->unsubscribe($fooConsumer);
+
         $context->createProducer()->send($fooQueue, $context->createMessage());
         $context->createProducer()->send($barQueue, $context->createMessage());
 
         $consumedMessages = 0;
-        $subscriptionConsumer->unsubscribe($fooConsumer);
         $subscriptionConsumer->consume(1000);
 
         $this->assertEquals(1, $consumedMessages);