fix: Mark channels closed when connection is closed

Author: Xfaider48

PhpAmqpLib/Channel/AMQPChannel.php

@@ -207,6 +207,11 @@ public function close($reply_code = 0, $reply_text = '', $method_sig = array(0,
         ), false, $this->channel_rpc_timeout);
     }
 
+    public function markClosed()
+    {
+        $this->do_close();
+    }
+
     /**
      * @param AMQPReader $reader
      * @throws AMQPProtocolChannelException

PhpAmqpLib/Connection/AbstractConnection.php

@@ -424,6 +424,7 @@ protected function do_close()
         $this->frame_queue = new \SplQueue();
         $this->method_queue = [];
         $this->setIsConnected(false);
+        $this->markChannelsClosed();
         $this->close_input();
         $this->close_socket();
     }
@@ -1112,6 +1113,21 @@ protected function closeChannels()
         }
     }
 
+    /**
+     * Mark all available channels as closed
+     */
+    protected function markChannelsClosed()
+    {
+        foreach ($this->channels as $key => $channel) {
+            // channels[0] is this connection object, so don't close it yet
+            if ($key === 0) {
+                continue;
+            }
+
+            $channel->markClosed();
+        }
+    }
+
     /**
      * Should the connection be attempted during construction?
      *

tests/Functional/Connection/ConnectionClosedTest.php

@@ -8,6 +8,7 @@
 use PhpAmqpLib\Exception\AMQPHeartbeatMissedException;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Tests\Functional\AbstractConnectionTest;
+use PhpAmqpLib\Tests\Functional\ToxiProxy;
 
 /**
  * @group connection
@@ -278,4 +279,82 @@ public function must_throw_exception_after_connection_was_restored($type)
         $this->assertChannelClosed($channel);
         $this->assertConnectionClosed($connection);
     }
+
+    /**
+     * Try to close and reopen connection with two channels.
+     *
+     * @test
+     * @small
+     * @group connection
+     * @group proxy
+     * @testWith ["stream"]
+     *            ["socket"]
+     * @covers \PhpAmqpLib\Wire\IO\StreamIO::write()
+     * @covers \PhpAmqpLib\Wire\IO\SocketIO::write()
+     *
+     * @param string $type
+     */
+    public function must_throw_exception_after_connection_was_restored_with_two_channels($type)
+    {
+        $timeout = 1;
+
+        // Create proxy part
+        $host = trim(getenv('TOXIPROXY_AMQP_TARGET'));
+        if (empty($host)) {
+            $host = HOST;
+        }
+        $proxy = new ToxiProxy('amqp_connection', $this->get_toxiproxy_host());
+        $proxy->open($host, PORT, $this->get_toxiproxy_amqp_port());
+
+        /** @var AbstractConnection $connection */
+        $connection = $this->connection_create(
+            $type,
+            $proxy->getHost(),
+            $proxy->getPort(),
+            array('timeout' => $timeout)
+        );
+
+        $channel = $connection->channel();
+        $anotherChannel = $connection->channel();
+
+        $this->assertTrue($channel->is_open());
+        $this->assertTrue($anotherChannel->is_open());
+
+        $this->queue_bind($channel, $exchange_name = 'test_exchange_broken', $queue_name);
+        $message = new AMQPMessage(
+            'test',
+            ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]
+        );
+        $channel->basic_publish($message, $exchange_name, $queue_name);
+
+        // drop proxy connection and wait longer than timeout
+        $proxy->close();
+        sleep($timeout);
+        usleep(100000);
+        // Reopen proxy
+        $proxy->open($host, PORT, $this->get_toxiproxy_amqp_port());
+
+        $retry = 0;
+        $exception = null;
+        do {
+            try {
+                $channel->basic_publish($message, $exchange_name, $queue_name);
+            } catch (\PHPUnit_Exception $exception) {
+                throw $exception;
+            } catch (\Exception $exception) {
+                break;
+            }
+        } while (!$exception && ++$retry < 100);
+
+        $this->assertInstanceOf(AMQPConnectionClosedException::class, $exception);
+        $this->assertGreaterThan(0, $exception->getCode());
+        $this->assertChannelClosed($channel);
+
+        // Now lets reconnect
+        $connection->reconnect();
+
+        // Both old channels must be closed
+        $this->assertChannelClosed($channel);
+        $this->assertChannelClosed($anotherChannel);
+    }
 }