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;
                     }
                 }