Skip to content

Commit a3c0c52

Browse files
appkit-frameworkWyriHaximus
appkit-framework
authored and
WyriHaximus
committed
Fix handling of exceptions thrown by frame handlers
1 parent d259b44 commit a3c0c52

3 files changed

Lines changed: 187 additions & 6 deletions

File tree

‎spec/generate.php‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@
55
namespace Bunny;
66

77
use Bunny\Protocol\ContentHeaderFrame;
8-
use Evenement\EventEmitterInterface;
9-
use Evenement\EventEmitterTrait;
108
use InvalidArgumentException;
119
use stdClass;
1210
use function array_filter;
@@ -288,9 +286,15 @@ function amqpTypeToLength(string $type, string $e): array
288286
$connectionContent .= " while ((\$frame = \$this->reader->consumeFrame(\$this->readBuffer)) !== null) {\n";
289287
$connectionContent .= " \$frameInAwaitList = false;\n";
290288
$connectionContent .= " foreach (\$this->awaitList as \$index => \$frameHandler) {\n";
291-
$connectionContent .= " if (\$frameHandler['filter'](\$frame)) {\n";
289+
$connectionContent .= " try {\n";
290+
$connectionContent .= " if (\$frameHandler['filter'](\$frame)) {\n";
291+
$connectionContent .= " unset(\$this->awaitList[\$index]);\n";
292+
$connectionContent .= " \$frameHandler['promise']->resolve(\$frame);\n";
293+
$connectionContent .= " \$frameInAwaitList = true;\n";
294+
$connectionContent .= " }\n";
295+
$connectionContent .= " } catch (Throwable \$exception) {\n";
292296
$connectionContent .= " unset(\$this->awaitList[\$index]);\n";
293-
$connectionContent .= " \$frameHandler['promise']->resolve(\$frame);\n";
297+
$connectionContent .= " \$frameHandler['promise']->reject(\$exception);\n";
294298
$connectionContent .= " \$frameInAwaitList = true;\n";
295299
$connectionContent .= " }\n";
296300
$connectionContent .= " }\n";

‎src/Connection.php‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,15 @@ public function __construct(
7070
while (($frame = $this->reader->consumeFrame($this->readBuffer)) !== null) {
7171
$frameInAwaitList = false;
7272
foreach ($this->awaitList as $index => $frameHandler) {
73-
if ($frameHandler['filter']($frame)) {
73+
try {
74+
if ($frameHandler['filter']($frame)) {
75+
unset($this->awaitList[$index]);
76+
$frameHandler['promise']->resolve($frame);
77+
$frameInAwaitList = true;
78+
}
79+
} catch (Throwable $exception) {
7480
unset($this->awaitList[$index]);
75-
$frameHandler['promise']->resolve($frame);
81+
$frameHandler['promise']->reject($exception);
7682
$frameInAwaitList = true;
7783
}
7884
}

‎test/ConnectionTest.php‎

Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,171 @@
1+
<?php
2+
3+
/** @noinspection PhpUnhandledExceptionInspection */
4+
5+
declare(strict_types=1);
6+
7+
namespace Bunny\Test;
8+
9+
use Bunny\Channels;
10+
use Bunny\Configuration;
11+
use Bunny\Connection;
12+
use Bunny\Constants;
13+
use Bunny\Exception\ClientException;
14+
use Bunny\Protocol\Buffer;
15+
use Bunny\Protocol\MethodConnectionCloseFrame;
16+
use Bunny\Protocol\ProtocolReader;
17+
use Bunny\Protocol\ProtocolWriter;
18+
use Bunny\Test\Library\ClientHelper;
19+
use Evenement\EventEmitterTrait;
20+
use PHPUnit\Framework\TestCase;
21+
use React\EventLoop\Loop;
22+
use React\EventLoop\StreamSelectLoop;
23+
use React\Promise\Deferred;
24+
use React\Socket\ConnectionInterface;
25+
use React\Stream\ThroughStream;
26+
use React\Stream\WritableStreamInterface;
27+
use RuntimeException;
28+
use Throwable;
29+
use WyriHaximus\React\PHPUnit\RunTestsInFibersTrait;
30+
use function React\Async\async;
31+
use function React\Async\await;
32+
33+
class ConnectionTest extends TestCase
34+
{
35+
use RunTestsInFibersTrait;
36+
37+
private ClientHelper $helper;
38+
39+
protected function setUp(): void
40+
{
41+
parent::setUp();
42+
43+
$this->helper = new ClientHelper();
44+
}
45+
46+
public function testThrowOn(): void
47+
{
48+
$oldLoop = Loop::get();
49+
Loop::set(new StreamSelectLoop());
50+
51+
self::expectException(ClientException::class);
52+
self::expectExceptionMessage('blaat');
53+
54+
// phpcs:disable
55+
$mockConnection = new class () implements ConnectionInterface {
56+
use EventEmitterTrait;
57+
58+
/**
59+
* @return string
60+
*/
61+
public function getRemoteAddress()
62+
{
63+
return '127.0.0.1:666';
64+
}
65+
66+
/**
67+
* @return string
68+
*/
69+
public function getLocalAddress()
70+
{
71+
return '127.0.0.1:666';
72+
}
73+
74+
public function isReadable()
75+
{
76+
return true;
77+
}
78+
79+
public function pause()
80+
{
81+
// No-op
82+
}
83+
84+
public function resume()
85+
{
86+
// No-op
87+
}
88+
89+
/**
90+
* @param array<mixed> $options
91+
*/
92+
public function pipe(WritableStreamInterface $dest, array $options = [])
93+
{
94+
return new ThroughStream();
95+
}
96+
97+
public function close()
98+
{
99+
// No-op
100+
}
101+
102+
/**
103+
* @retrun bool
104+
*/
105+
public function isWritable()
106+
{
107+
return true;
108+
}
109+
110+
/**
111+
* @param string $data
112+
*
113+
* @retrun bool
114+
*/
115+
public function write($data)
116+
{
117+
return false;
118+
}
119+
120+
/**
121+
* @param ?string $data
122+
*/
123+
public function end($data = null)
124+
{
125+
// No-op
126+
}
127+
};
128+
// phpcs:enable
129+
$buffer = new Buffer();
130+
$frame = new MethodConnectionCloseFrame();
131+
$frame->replyCode = Constants::STATUS_REPLY_SUCCESS;
132+
$frame->replyText = 'blaat';
133+
$frame->closeClassId = Constants::CLASS_CONNECTION;
134+
$frame->closeMethodId = Constants::METHOD_CONNECTION_CLOSE;
135+
(new ProtocolWriter())->appendFrame($frame, $buffer);
136+
$connection = new Connection(
137+
$this->helper->createClient(),
138+
$mockConnection,
139+
new Buffer(),
140+
new Buffer(),
141+
new ProtocolReader(),
142+
new ProtocolWriter(),
143+
new Channels(),
144+
new Configuration(),
145+
);
146+
$deferred = new Deferred();
147+
Loop::addTimer(0.1, async(static function () use ($deferred, $connection): void {
148+
try {
149+
$connection->awaitAck(666);
150+
$deferred->reject(new RuntimeException('We should not reach this line'));
151+
} catch (Throwable $exception) {
152+
$deferred->reject($exception);
153+
}
154+
}));
155+
Loop::addTimer(0.2, async(static function () use ($mockConnection, $buffer): void {
156+
$line = $buffer->consume($buffer->getLength());
157+
$mockConnection->emit('data', [$line]);
158+
}));
159+
Loop::addTimer(0.3, async(static function () use ($mockConnection): void {
160+
$mockConnection->emit('drain');
161+
}));
162+
Loop::addTimer(1, async(static function (): void {
163+
Loop::stop();
164+
}));
165+
166+
Loop::run();
167+
Loop::set($oldLoop);
168+
169+
await($deferred->promise());
170+
}
171+
}

0 commit comments

Comments
 (0)