Skip to content

Commit d46e3fe

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

3 files changed

Lines changed: 187 additions & 6 deletions

File tree

‎spec/generate.php‎

Lines changed: 9 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,10 +286,17 @@ 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";
299+
//$connectionContent .= " return;\n";
295300
$connectionContent .= " }\n";
296301
$connectionContent .= " }\n";
297302
$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: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
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+
53+
// phpcs:disable
54+
$mockConnection = new class () implements ConnectionInterface {
55+
use EventEmitterTrait;
56+
57+
/**
58+
* @return string
59+
*/
60+
public function getRemoteAddress()
61+
{
62+
return '127.0.0.1:666';
63+
}
64+
65+
/**
66+
* @return string
67+
*/
68+
public function getLocalAddress()
69+
{
70+
return '127.0.0.1:666';
71+
}
72+
73+
public function isReadable()
74+
{
75+
return true;
76+
}
77+
78+
public function pause()
79+
{
80+
// No-op
81+
}
82+
83+
public function resume()
84+
{
85+
// No-op
86+
}
87+
88+
/**
89+
* @param array<mixed> $options
90+
*/
91+
public function pipe(WritableStreamInterface $dest, array $options = [])
92+
{
93+
return new ThroughStream();
94+
}
95+
96+
public function close()
97+
{
98+
// No-op
99+
}
100+
101+
/**
102+
* @retrun bool
103+
*/
104+
public function isWritable()
105+
{
106+
return true;
107+
}
108+
109+
/**
110+
* @param string $data
111+
*
112+
* @retrun bool
113+
*/
114+
public function write($data)
115+
{
116+
return false;
117+
}
118+
119+
/**
120+
* @param ?string $data
121+
*/
122+
public function end($data = null)
123+
{
124+
// No-op
125+
}
126+
};
127+
// phpcs:enable
128+
$buffer = new Buffer();
129+
$frame = new MethodConnectionCloseFrame();
130+
$frame->replyCode = Constants::STATUS_REPLY_SUCCESS;
131+
$frame->replyText = 'blaat';
132+
$frame->closeClassId = Constants::CLASS_CONNECTION;
133+
$frame->closeMethodId = Constants::METHOD_CONNECTION_CLOSE;
134+
(new ProtocolWriter())->appendFrame($frame, $buffer);
135+
$connection = new Connection(
136+
$this->helper->createClient(),
137+
$mockConnection,
138+
new Buffer(),
139+
new Buffer(),
140+
new ProtocolReader(),
141+
new ProtocolWriter(),
142+
new Channels(),
143+
new Configuration(),
144+
);
145+
$deferred = new Deferred();
146+
Loop::addTimer(0.1, async(static function () use ($deferred, $connection): void {
147+
try {
148+
$connection->awaitAck(666);
149+
$deferred->reject(new RuntimeException('We should not reach this line'));
150+
} catch (Throwable $exception) {
151+
$deferred->reject($exception);
152+
}
153+
}));
154+
Loop::addTimer(0.2, async(static function () use ($mockConnection, $buffer): void {
155+
$line = $buffer->consume($buffer->getLength());
156+
$mockConnection->emit('data', [$line]);
157+
}));
158+
Loop::addTimer(0.3, async(static function () use ($mockConnection): void {
159+
$mockConnection->emit('drain');
160+
}));
161+
Loop::addTimer(1, async(static function (): void {
162+
Loop::stop();
163+
}));
164+
165+
Loop::run();
166+
Loop::set($oldLoop);
167+
168+
await($deferred->promise());
169+
}
170+
}

0 commit comments

Comments
 (0)