[WIP] Disconnect and emit error on unrecoverable errors

composer.json

@@ -27,7 +27,8 @@
     "require": {
         "php": "^7.1 || ^8.0",
         "react/event-loop": "^1.0 || ^0.5 || ^0.4",
-        "react/promise": "~2.2"
+        "react/promise": "^3 || ~2.2",
+        "evenement/evenement": "^3.0"
     },
     "require-dev": {
         "ext-pcntl": "*",

src/Bunny/AbstractClient.php

@@ -13,6 +13,7 @@
 use Bunny\Protocol\MethodFrame;
 use Bunny\Protocol\ProtocolReader;
 use Bunny\Protocol\ProtocolWriter;
+use Evenement\EventEmitter;
 use InvalidArgumentException;
 use React\Promise;
 
@@ -29,7 +30,7 @@
  *
  * @author Jakub Kulhan <jakub.kulhan@gmail.com>
  */
-abstract class AbstractClient
+abstract class AbstractClient extends EventEmitter
 {
 
     use ClientMethods;

src/Bunny/Async/Client.php

@@ -299,26 +299,31 @@ public function onDataAvailable()
     {
         $this->read();
 
-        while (($frame = $this->reader->consumeFrame($this->readBuffer)) !== null) {
-            foreach ($this->awaitCallbacks as $k => $callback) {
-                if ($callback($frame) === true) {
-                    unset($this->awaitCallbacks[$k]);
-                    continue 2; // CONTINUE WHILE LOOP
+        try {
+            while (($frame = $this->reader->consumeFrame($this->readBuffer)) !== null) {
+                foreach ($this->awaitCallbacks as $k => $callback) {
+                    if ($callback($frame) === true) {
+                        unset($this->awaitCallbacks[$k]);
+                        continue 2; // CONTINUE WHILE LOOP
+                    }
                 }
-            }
 
-            if ($frame->channel === 0) {
-                $this->onFrameReceived($frame);
+                if ($frame->channel === 0) {
+                    $this->onFrameReceived($frame);
 
-            } else {
-                if (!isset($this->channels[$frame->channel])) {
-                    throw new ClientException(
-                        "Received frame #{$frame->type} on closed channel #{$frame->channel}."
-                    );
-                }
+                } else {
+                    if (!isset($this->channels[$frame->channel])) {
+                        throw new ClientException(
+                            "Received frame #{$frame->type} on closed channel #{$frame->channel}."
+                        );
+                    }
 
-                $this->channels[$frame->channel]->onFrameReceived($frame);
+                    $this->channels[$frame->channel]->onFrameReceived($frame);
+                }
             }
+        } catch (\Throwable $error) {
+            $this->disconnect();
+            $this->emit('error', [$error]);
         }
     }