Skip to content

Commit 8258870

Browse files
WyriHaximusWyriHaximus
authored
Merge pull request #200 from appkit-framework/emit-error-events
Emit 'error' events instead of throwing uncatchable exceptions
2 parents dcee94b + 4e93c7d commit 8258870

6 files changed

Lines changed: 206 additions & 136 deletions

File tree

‎spec/generate.php‎

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

77
use Bunny\Protocol\ContentHeaderFrame;
8+
use Evenement\EventEmitterInterface;
9+
use Evenement\EventEmitterTrait;
810
use InvalidArgumentException;
911
use stdClass;
1012
use function array_filter;
@@ -230,11 +232,14 @@ function amqpTypeToLength(string $type, string $e): array
230232
$connectionContent .= "use Bunny\\Protocol\\MethodFrame;\n";
231233
$connectionContent .= "use Bunny\\Protocol\\ProtocolReader;\n";
232234
$connectionContent .= "use Bunny\\Protocol\\ProtocolWriter;\n";
235+
$connectionContent .= "use Evenement\\EventEmitterInterface;\n";
236+
$connectionContent .= "use Evenement\\EventEmitterTrait;\n";
233237
$connectionContent .= "use React\\EventLoop\\Loop;\n";
234238
$connectionContent .= "use React\\EventLoop\\TimerInterface;\n";
235239
$connectionContent .= "use React\\Promise\\Deferred;\n";
236240
$connectionContent .= "use React\\Promise\\Promise;\n";
237241
$connectionContent .= "use React\\Socket\\ConnectionInterface;\n";
242+
$connectionContent .= "use Throwable;\n";
238243
$connectionContent .= "use function React\\Async\\await;\n";
239244
$connectionContent .= "use function count;\n";
240245
$connectionContent .= "use function is_callable;\n";
@@ -253,8 +258,10 @@ function amqpTypeToLength(string $type, string $e): array
253258
$connectionContent .= " *\n";
254259
$connectionContent .= " * @author Jakub Kulhan <jakub.kulhan@gmail.com>\n";
255260
$connectionContent .= " */\n";
256-
$connectionContent .= "final class Connection\n";
261+
$connectionContent .= "final class Connection implements EventEmitterInterface\n";
257262
$connectionContent .= "{\n";
263+
$connectionContent .= " use EventEmitterTrait;\n";
264+
$connectionContent .= "\n";
258265
$connectionContent .= " protected ?TimerInterface \$heartbeatTimer = null;\n";
259266
$connectionContent .= "\n";
260267
$connectionContent .= " protected float \$lastWrite = 0.0;\n";
@@ -292,16 +299,20 @@ function amqpTypeToLength(string $type, string $e): array
292299
$connectionContent .= " continue;\n";
293300
$connectionContent .= " }\n";
294301
$connectionContent .= "\n";
295-
$connectionContent .= " if (\$frame->channel === 0) {\n";
296-
$connectionContent .= " \$this->onFrameReceived(\$frame);\n";
297-
$connectionContent .= " continue;\n";
298-
$connectionContent .= " }\n";
302+
$connectionContent .= " try {\n";
303+
$connectionContent .= " if (\$frame->channel === 0) {\n";
304+
$connectionContent .= " \$this->onFrameReceived(\$frame);\n";
305+
$connectionContent .= " continue;\n";
306+
$connectionContent .= " }\n";
299307
$connectionContent .= "\n";
300-
$connectionContent .= " if (!\$this->channels->has(\$frame->channel)) {\n";
301-
$connectionContent .= " throw new ClientException(sprintf('Received frame #%d on closed channel #%d.', \$frame->type, \$frame->channel));\n";
302-
$connectionContent .= " }\n";
308+
$connectionContent .= " if (!\$this->channels->has(\$frame->channel)) {\n";
309+
$connectionContent .= " throw new ClientException(sprintf('Received frame #%d on closed channel #%d.', \$frame->type, \$frame->channel));\n";
310+
$connectionContent .= " }\n";
303311
$connectionContent .= "\n";
304-
$connectionContent .= " \$this->channels->get(\$frame->channel)->onFrameReceived(\$frame);\n";
312+
$connectionContent .= " \$this->channels->get(\$frame->channel)->onFrameReceived(\$frame);\n";
313+
$connectionContent .= " } catch (Throwable \$e) {\n";
314+
$connectionContent .= " \$this->emit('error', [\$e]);\n";
315+
$connectionContent .= " }\n";
305316
$connectionContent .= " }\n";
306317
$connectionContent .= " });\n";
307318
$connectionContent .= " }\n";
@@ -1029,14 +1040,14 @@ function amqpTypeToLength(string $type, string $e): array
10291040

10301041
$connectionContent .= " public function startHeartbeatTimer(): void\n";
10311042
$connectionContent .= " {\n";
1032-
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$this->configuration->heartbeat, [\$this, 'onHeartbeat']);\n";
1033-
$connectionContent .= " \$this->connection->on('drain', [\$this, 'onHeartbeat']);\n";
1043+
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$this->configuration->heartbeat, \$this->onHeartbeat(...));\n";
1044+
$connectionContent .= " \$this->connection->on('drain', \$this->onHeartbeat(...));\n";
10341045
$connectionContent .= " }\n";
10351046
$connectionContent .= "\n";
10361047
$connectionContent .= " /**\n";
10371048
$connectionContent .= " * Callback when heartbeat timer timed out.\n";
10381049
$connectionContent .= " */\n";
1039-
$connectionContent .= " public function onHeartbeat(): void\n";
1050+
$connectionContent .= " private function onHeartbeat(): void\n";
10401051
$connectionContent .= " {\n";
10411052
$connectionContent .= " \$now = microtime(true);\n";
10421053
$connectionContent .= " \$nextHeartbeat = (\$this->lastWrite ?: \$now) + \$this->configuration->heartbeat;\n";
@@ -1045,12 +1056,12 @@ function amqpTypeToLength(string $type, string $e): array
10451056
$connectionContent .= " \$this->writer->appendFrame(new HeartbeatFrame(), \$this->writeBuffer);\n";
10461057
$connectionContent .= " \$this->flushWriteBuffer();\n";
10471058
$connectionContent .= "\n";
1048-
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$this->configuration->heartbeat, [\$this, 'onHeartbeat']);\n";
1059+
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$this->configuration->heartbeat, \$this->onHeartbeat(...));\n";
10491060
$connectionContent .= " if (is_callable(\$this->configuration->heartbeatCallback)) {\n";
10501061
$connectionContent .= " (\$this->configuration->heartbeatCallback)(\$this);\n";
10511062
$connectionContent .= " }\n";
10521063
$connectionContent .= " } else {\n";
1053-
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$nextHeartbeat - \$now, [\$this, 'onHeartbeat']);\n";
1064+
$connectionContent .= " \$this->heartbeatTimer = Loop::addTimer(\$nextHeartbeat - \$now, \$this->onHeartbeat(...));\n";
10541065
$connectionContent .= " }\n";
10551066
$connectionContent .= " }\n";
10561067
$connectionContent .= "}\n";

‎src/Channel.php‎

Lines changed: 115 additions & 111 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,12 @@
2525
use Bunny\Protocol\MethodTxCommitOkFrame;
2626
use Bunny\Protocol\MethodTxRollbackOkFrame;
2727
use Bunny\Protocol\MethodTxSelectOkFrame;
28-
use Evenement\EventEmitterInterface;
2928
use Evenement\EventEmitterTrait;
3029
use LogicException;
3130
use React\Promise\Deferred;
3231
use React\Promise\PromiseInterface;
3332
use SplQueue;
33+
use Throwable;
3434
use function React\Async\async;
3535
use function React\Async\await;
3636
use function array_key_exists;
@@ -47,7 +47,7 @@
4747
* @author Jakub Kulhan <jakub.kulhan@gmail.com>
4848
* @final Will be marked final in a future major release
4949
*/
50-
class Channel implements ChannelInterface, EventEmitterInterface
50+
class Channel implements ChannelInterface
5151
{
5252
use EventEmitterTrait;
5353
use ChannelMethods {
@@ -438,141 +438,145 @@ private function enterConfirmMode(?callable $callback = null): void
438438
*/
439439
public function onFrameReceived(AbstractFrame $frame): void
440440
{
441-
if ($this->state === ChannelState::Error) {
442-
throw new ChannelException('Channel in error state.');
443-
}
444-
445-
if ($this->state === ChannelState::Closed) {
446-
throw new ChannelException(sprintf('Received frame #%d on closed channel #%d.', $frame->type, $this->channelId));
447-
}
448-
449-
if ($frame instanceof MethodFrame) {
450-
if ($this->state === ChannelState::Closing && !($frame instanceof MethodChannelCloseOkFrame)) {
451-
// drop frames in closing state
452-
return;
441+
try {
442+
if ($this->state === ChannelState::Error) {
443+
throw new ChannelException('Channel in error state.');
453444
}
454445

455-
if ($this->state !== ChannelState::Ready && !($frame instanceof MethodChannelCloseOkFrame)) {
456-
$currentState = $this->state;
457-
$this->state = ChannelState::Error;
446+
if ($this->state === ChannelState::Closed) {
447+
throw new ChannelException(sprintf('Received frame #%d on closed channel #%d.', $frame->type, $this->channelId));
448+
}
458449

459-
if ($currentState === ChannelState::AwaitingHeader) {
460-
$msg = 'Got method frame, expected header frame.';
461-
} elseif ($currentState === ChannelState::AwaitingBody) {
462-
$msg = 'Got method frame, expected body frame.';
463-
} else {
464-
throw new LogicException('Unhandled channel state.');
450+
if ($frame instanceof MethodFrame) {
451+
if ($this->state === ChannelState::Closing && !($frame instanceof MethodChannelCloseOkFrame)) {
452+
// drop frames in closing state
453+
return;
465454
}
466455

467-
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
456+
if ($this->state !== ChannelState::Ready && !($frame instanceof MethodChannelCloseOkFrame)) {
457+
$currentState = $this->state;
458+
$this->state = ChannelState::Error;
468459

469-
throw new ChannelException('Unexpected frame: ' . $msg);
470-
}
460+
if ($currentState === ChannelState::AwaitingHeader) {
461+
$msg = 'Got method frame, expected header frame.';
462+
} elseif ($currentState === ChannelState::AwaitingBody) {
463+
$msg = 'Got method frame, expected body frame.';
464+
} else {
465+
throw new LogicException('Unhandled channel state.');
466+
}
471467

472-
if ($frame instanceof MethodChannelCloseOkFrame) {
473-
$this->state = ChannelState::Closed;
468+
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
474469

475-
if ($this->closeDeferred !== null) {
476-
$this->closeDeferred->resolve($this->channelId);
470+
throw new ChannelException('Unexpected frame: ' . $msg);
477471
}
478472

479-
// // break reference cycle, must be called after resolving promise
480-
// $this->client = null;
481-
// break consumers' reference cycle
482-
$this->consumeConcurrency = [];
483-
$this->consumeConcurrent = [];
484-
$this->deliverCallbacks = [];
485-
$this->deliveryQueue = [];
486-
} elseif ($frame instanceof MethodBasicReturnFrame) {
487-
$this->returnFrame = $frame;
488-
$this->state = ChannelState::AwaitingHeader;
489-
} elseif ($frame instanceof MethodBasicDeliverFrame) {
490-
$this->deliverFrame = $frame;
491-
$this->state = ChannelState::AwaitingHeader;
492-
} elseif ($frame instanceof MethodBasicAckFrame) {
493-
foreach ($this->ackCallbacks as $callback) {
494-
$callback($frame);
473+
if ($frame instanceof MethodChannelCloseOkFrame) {
474+
$this->state = ChannelState::Closed;
475+
476+
if ($this->closeDeferred !== null) {
477+
$this->closeDeferred->resolve($this->channelId);
478+
}
479+
480+
// break reference cycle, must be called after resolving promise
481+
// $this->client = null;
482+
// break consumers' reference cycle
483+
$this->consumeConcurrency = [];
484+
$this->consumeConcurrent = [];
485+
$this->deliverCallbacks = [];
486+
$this->deliveryQueue = [];
487+
} elseif ($frame instanceof MethodBasicReturnFrame) {
488+
$this->returnFrame = $frame;
489+
$this->state = ChannelState::AwaitingHeader;
490+
} elseif ($frame instanceof MethodBasicDeliverFrame) {
491+
$this->deliverFrame = $frame;
492+
$this->state = ChannelState::AwaitingHeader;
493+
} elseif ($frame instanceof MethodBasicAckFrame) {
494+
foreach ($this->ackCallbacks as $callback) {
495+
$callback($frame);
496+
}
497+
} elseif ($frame instanceof MethodBasicNackFrame) {
498+
foreach ($this->ackCallbacks as $callback) {
499+
$callback($frame);
500+
}
501+
} elseif ($frame instanceof MethodChannelCloseFrame) {
502+
throw new ChannelException('Channel closed by server: ' . $frame->replyText, $frame->replyCode);
503+
} else {
504+
throw new ChannelException('Unhandled method frame ' . $frame::class . '.');
495505
}
496-
} elseif ($frame instanceof MethodBasicNackFrame) {
497-
foreach ($this->ackCallbacks as $callback) {
498-
$callback($frame);
506+
} elseif ($frame instanceof ContentHeaderFrame) {
507+
if ($this->state === ChannelState::Closing) {
508+
// drop frames in closing state
509+
return;
499510
}
500-
} elseif ($frame instanceof MethodChannelCloseFrame) {
501-
throw new ChannelException('Channel closed by server: ' . $frame->replyText, $frame->replyCode);
502-
} else {
503-
throw new ChannelException('Unhandled method frame ' . $frame::class . '.');
504-
}
505-
} elseif ($frame instanceof ContentHeaderFrame) {
506-
if ($this->state === ChannelState::Closing) {
507-
// drop frames in closing state
508-
return;
509-
}
510511

511-
if ($this->state !== ChannelState::AwaitingHeader) {
512-
$currentState = $this->state;
513-
$this->state = ChannelState::Error;
512+
if ($this->state !== ChannelState::AwaitingHeader) {
513+
$currentState = $this->state;
514+
$this->state = ChannelState::Error;
515+
516+
if ($currentState === ChannelState::Ready) {
517+
$msg = 'Got header frame, expected method frame.';
518+
} elseif ($currentState === ChannelState::AwaitingBody) {
519+
$msg = 'Got header frame, expected content frame.';
520+
} else {
521+
throw new LogicException('Unhandled channel state.');
522+
}
514523

515-
if ($currentState === ChannelState::Ready) {
516-
$msg = 'Got header frame, expected method frame.';
517-
} elseif ($currentState === ChannelState::AwaitingBody) {
518-
$msg = 'Got header frame, expected content frame.';
519-
} else {
520-
throw new LogicException('Unhandled channel state.');
524+
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
525+
526+
throw new ChannelException('Unexpected frame: ' . $msg);
521527
}
522528

523-
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
529+
$this->headerFrame = $frame;
530+
$this->bodySizeRemaining = $frame->bodySize;
524531

525-
throw new ChannelException('Unexpected frame: ' . $msg);
526-
}
532+
if ($this->bodySizeRemaining > 0) {
533+
$this->state = ChannelState::AwaitingBody;
534+
} else {
535+
$this->state = ChannelState::Ready;
536+
$this->onBodyComplete();
537+
}
538+
} elseif ($frame instanceof ContentBodyFrame) {
539+
if ($this->state === ChannelState::Closing) {
540+
// drop frames in closing state
541+
return;
542+
}
527543

528-
$this->headerFrame = $frame;
529-
$this->bodySizeRemaining = $frame->bodySize;
544+
if ($this->state !== ChannelState::AwaitingBody) {
545+
$currentState = $this->state;
546+
$this->state = ChannelState::Error;
530547

531-
if ($this->bodySizeRemaining > 0) {
532-
$this->state = ChannelState::AwaitingBody;
533-
} else {
534-
$this->state = ChannelState::Ready;
535-
$this->onBodyComplete();
536-
}
537-
} elseif ($frame instanceof ContentBodyFrame) {
538-
if ($this->state === ChannelState::Closing) {
539-
// drop frames in closing state
540-
return;
541-
}
548+
if ($currentState === ChannelState::Ready) {
549+
$msg = 'Got body frame, expected method frame.';
550+
} elseif ($currentState === ChannelState::AwaitingHeader) {
551+
$msg = 'Got body frame, expected header frame.';
552+
} else {
553+
throw new LogicException('Unhandled channel state.');
554+
}
542555

543-
if ($this->state !== ChannelState::AwaitingBody) {
544-
$currentState = $this->state;
545-
$this->state = ChannelState::Error;
556+
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
546557

547-
if ($currentState === ChannelState::Ready) {
548-
$msg = 'Got body frame, expected method frame.';
549-
} elseif ($currentState === ChannelState::AwaitingHeader) {
550-
$msg = 'Got body frame, expected header frame.';
551-
} else {
552-
throw new LogicException('Unhandled channel state.');
558+
throw new ChannelException('Unexpected frame: ' . $msg);
553559
}
554560

555-
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, $msg);
556-
557-
throw new ChannelException('Unexpected frame: ' . $msg);
558-
}
561+
$this->bodyBuffer->append($frame->payload);
562+
$this->bodySizeRemaining -= $frame->payloadSize;
559563

560-
$this->bodyBuffer->append($frame->payload);
561-
$this->bodySizeRemaining -= $frame->payloadSize;
564+
if ($this->bodySizeRemaining < 0) {
565+
$this->state = ChannelState::Error;
566+
$this->connection->disconnect(Constants::STATUS_SYNTAX_ERROR, 'Body overflow, received ' . (-$this->bodySizeRemaining) . ' more bytes.');
567+
} elseif ($this->bodySizeRemaining === 0) {
568+
$this->state = ChannelState::Ready;
569+
$this->onBodyComplete();
570+
}
571+
} elseif ($frame instanceof HeartbeatFrame) {
572+
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, 'Got heartbeat on non-zero channel.');
562573

563-
if ($this->bodySizeRemaining < 0) {
564-
$this->state = ChannelState::Error;
565-
$this->connection->disconnect(Constants::STATUS_SYNTAX_ERROR, 'Body overflow, received ' . (-$this->bodySizeRemaining) . ' more bytes.');
566-
} elseif ($this->bodySizeRemaining === 0) {
567-
$this->state = ChannelState::Ready;
568-
$this->onBodyComplete();
574+
throw new ChannelException('Unexpected heartbeat frame.');
575+
} else {
576+
throw new ChannelException('Unhandled frame ' . $frame::class . '.');
569577
}
570-
} elseif ($frame instanceof HeartbeatFrame) {
571-
$this->connection->disconnect(Constants::STATUS_UNEXPECTED_FRAME, 'Got heartbeat on non-zero channel.');
572-
573-
throw new ChannelException('Unexpected heartbeat frame.');
574-
} else {
575-
throw new ChannelException('Unhandled frame ' . $frame::class . '.');
578+
} catch (Throwable $e) {
579+
$this->emit('error', [$e]);
576580
}
577581
}
578582

‎src/ChannelInterface.php‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
use Bunny\Protocol\MethodTxCommitOkFrame;
2222
use Bunny\Protocol\MethodTxRollbackOkFrame;
2323
use Bunny\Protocol\MethodTxSelectOkFrame;
24+
use Evenement\EventEmitterInterface;
2425

2526
/**
2627
* AMQP channel.
@@ -31,7 +32,7 @@
3132
* @author Jakub Kulhan <jakub.kulhan@gmail.com>
3233
* @final Will be marked final in a future major release
3334
*/
34-
interface ChannelInterface
35+
interface ChannelInterface extends EventEmitterInterface
3536
{
3637
/**
3738
* Returns the channel mode.

0 commit comments

Comments
 (0)