Fix handling of exceptions thrown by frame handlers

Author: appkit-framework

spec/generate.php

@@ -288,9 +288,15 @@ function amqpTypeToLength(string $type, string $e): array
 $connectionContent .= "            while ((\$frame = \$this->reader->consumeFrame(\$this->readBuffer)) !== null) {\n";
 $connectionContent .= "                \$frameInAwaitList = false;\n";
 $connectionContent .= "                foreach (\$this->awaitList as \$index => \$frameHandler) {\n";
-$connectionContent .= "                    if (\$frameHandler['filter'](\$frame)) {\n";
+$connectionContent .= "                    try {\n";
+$connectionContent .= "                        if (\$frameHandler['filter'](\$frame)) {\n";
+$connectionContent .= "                            unset(\$this->awaitList[\$index]);\n";
+$connectionContent .= "                            \$frameHandler['promise']->resolve(\$frame);\n";
+$connectionContent .= "                            \$frameInAwaitList = true;\n";
+$connectionContent .= "                        }\n";
+$connectionContent .= "                    } catch (Throwable \$exception) {\n";
 $connectionContent .= "                        unset(\$this->awaitList[\$index]);\n";
-$connectionContent .= "                        \$frameHandler['promise']->resolve(\$frame);\n";
+$connectionContent .= "                        \$frameHandler['promise']->reject(\$exception);\n";
 $connectionContent .= "                        \$frameInAwaitList = true;\n";
 $connectionContent .= "                    }\n";
 $connectionContent .= "                }\n";

src/Connection.php

@@ -70,9 +70,15 @@ public function __construct(
             while (($frame = $this->reader->consumeFrame($this->readBuffer)) !== null) {
                 $frameInAwaitList = false;
                 foreach ($this->awaitList as $index => $frameHandler) {
-                    if ($frameHandler['filter']($frame)) {
+                    try {
+                        if ($frameHandler['filter']($frame)) {
+                            unset($this->awaitList[$index]);
+                            $frameHandler['promise']->resolve($frame);
+                            $frameInAwaitList = true;
+                        }
+                    } catch (Throwable $exception) {
                         unset($this->awaitList[$index]);
-                        $frameHandler['promise']->resolve($frame);
+                        $frameHandler['promise']->reject($exception);
                         $frameInAwaitList = true;
                     }
                 }

test/ConnectionTest.php

@@ -0,0 +1,129 @@
+<?php
+
+/** @noinspection PhpUnhandledExceptionInspection */
+
+declare(strict_types=1);
+
+namespace Bunny\Test;
+
+use Bunny\Channels;
+use Bunny\Configuration;
+use Bunny\Connection;
+use Bunny\Protocol\Buffer;
+use Bunny\Protocol\ProtocolReader;
+use Bunny\Protocol\ProtocolWriter;
+use Bunny\Test\Library\ClientHelper;
+use Evenement\EventEmitterTrait;
+use PHPUnit\Framework\TestCase;
+use React\EventLoop\Loop;
+use React\Promise\Deferred;
+use React\Socket\ConnectionInterface;
+use React\Stream\WritableStreamInterface;
+use WyriHaximus\React\PHPUnit\RunTestsInFibersTrait;
+use function React\Async\await;
+
+class ConnectionTest extends TestCase
+{
+    use RunTestsInFibersTrait;
+
+    private ClientHelper $helper;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        $this->helper = new ClientHelper();
+    }
+
+    public function testThrowOn(): void
+    {
+        // phpcs:disable
+        $mockConnection = new class () implements ConnectionInterface {
+            use EventEmitterTrait;
+
+            /**
+             * @return string
+             */
+            public function getRemoteAddress()
+            {
+                return '127.0.0.1:666';
+            }
+
+            /**
+             * @return string
+             */
+            public function getLocalAddress()
+            {
+                return '127.0.0.1:666';
+            }
+
+            public function isReadable(): void
+            {
+                // TODO: Implement isReadable() method.
+            }
+
+            public function pause(): void
+            {
+                // TODO: Implement pause() method.
+            }
+
+            public function resume(): void
+            {
+                // TODO: Implement resume() method.
+            }
+
+            /**
+             * @param array $options
+             */
+            public function pipe(WritableStreamInterface $dest, array $options = []): void
+            {
+                // TODO: Implement pipe() method.
+            }
+
+            public function close(): void
+            {
+                // TODO: Implement close() method.
+            }
+
+            public function isWritable(): void
+            {
+                // TODO: Implement isWritable() method.
+            }
+
+            /**
+             * @param string $data
+             */
+            public function write($data): void
+            {
+                // TODO: Implement write() method.
+            }
+
+            /**
+             * @param ?string $data
+             */
+            public function end($data = null): void
+            {
+                // TODO: Implement end() method.
+            }
+        };
+        // phpcs:enable
+        $connection = new Connection(
+            $this->helper->createClient(),
+            $mockConnection,
+            new Buffer(),
+            new Buffer(),
+            new ProtocolReader(),
+            new ProtocolWriter(),
+            new Channels(),
+            new Configuration(),
+        );
+        $mockConnection->emit(
+            'data',
+            [
+                ''
+            ],
+        );
+//        $connection->awaitAck(666);
+//        $connection->awaitContentHeader(666);
+    }
+}