[0.6.x] Handle channel closing mid handling a message

Found this while working on #193

Author: WyriHaximus

src/Channel.php

@@ -636,6 +636,15 @@ private function deliveryTick(string $consumerTag): void
 
         $outcome = $callback($message, $this, $this->client);
 
+        /**
+         * Handle channel close getting called while handling a message.
+         * Also not handling more messages when the channel is closing or when it errors.
+         * Those put the channel in a state we can't consume any more messages.
+         */
+        if ($this->state === ChannelState::Closing || $this->state === ChannelState::Closed || $this->state === ChannelState::Error) {
+            return;
+        }
+
         if (!($outcome instanceof PromiseInterface)) {
             $this->consumeConcurrent[$consumerTag]--;
             $this->deliveryTick($consumerTag);

test/ChannelTest.php

@@ -242,4 +242,27 @@ public function testBigMessage(): void
         $c->disconnect();
         self::assertFalse($c->isConnected());
     }
+
+    public function testChannelClosingMidMessageHandling(): void
+    {
+        /**
+         * @var Deferred<bool> $deferred
+         */
+        $deferred = new Deferred();
+        $body = 'abc';
+
+        $c = $this->helper->createClient();
+
+        $ch = $c->connect()->channel();
+        $ch->queueDeclare('test_queue', false, false, false, true);
+        $ch->consume(static function (Message $msg, Channel $ch, Client $c) use ($body, $deferred): void {
+            self::assertEquals($body, $msg->content);
+            $deferred->resolve(true);
+            $c->disconnect();
+        });
+        $ch->publish($body, [], '', 'test_queue');
+        self::assertTrue(await($deferred->promise()));
+
+        self::assertFalse($c->isConnected());
+    }
 }