Added connection monitoring

Author: mmasiukevich

examples/basic.php

@@ -28,7 +28,7 @@
     }
 
     yield $channel->consume(function (Message $message, Channel $channel) {
-        echo $message->content() . \PHP_EOL;
+        echo $message->content . \PHP_EOL;
 
         yield $channel->ack($message);
     }, 'basic_queue');

examples/worker.php

@@ -32,7 +32,7 @@
     echo '[*] Waiting for messages. To exit press CTRL+C', \PHP_EOL;
 
     $tag = yield $channel->consume(function (Message $message, Channel $channel) {
-        echo "[x] Received message: {$message->content()}.", \PHP_EOL;
+        echo "[x] Received message: {$message->content}.", \PHP_EOL;
 
         // Do some work - we generate password hashes with a high cost
         // sleep() gets interrupted by Ctrl+C so it's not very good for demos
@@ -43,12 +43,12 @@
             password_hash(random_bytes(255), PASSWORD_BCRYPT, ["cost" => 15]);
         }
 
-        echo "[x] Done ", $message->content(), \PHP_EOL;
+        echo "[x] Done ", $message->content, \PHP_EOL;
 
         try {
             yield $channel->ack($message);
 
-            echo "ACK SUCCESS:: {$message->content()}", \PHP_EOL;
+            echo "ACK SUCCESS:: {$message->content}", \PHP_EOL;
         } catch (\Throwable $error) {
 
             echo "ACK FAILED:: {$error->getMessage()}", \PHP_EOL;

src/Client.php

@@ -12,6 +12,7 @@
 
 namespace PHPinnacle\Ridge;
 
+use Amp\Loop;
 use function Amp\asyncCall;
 use function Amp\call;
 use Amp\Deferred;
@@ -24,6 +25,8 @@ final class Client
     private const STATE_CONNECTED = 2;
     private const STATE_DISCONNECTING = 3;
 
+    private const CONNECTION_MONITOR_INTERVAL = 5000;
+
     /**
      * @var Config
      */
@@ -54,6 +57,11 @@ final class Client
      */
     private $properties;
 
+    /**
+     * @var string|null
+     */
+    private $connectionMonitorWatcherId;
+
     public function __construct(Config $config)
     {
         $this->config = $config;
@@ -91,7 +99,7 @@ function () {
 
                 $this->state = self::STATE_CONNECTING;
 
-                $this->connection = new Connection($this->config->uri(), fn() => $this->state = self::STATE_NOT_CONNECTED);
+                $this->connection = new Connection($this->config->uri());
 
                 yield $this->connection->open(
                     $this->config->timeout,
@@ -132,6 +140,16 @@ function () {
                 );
 
                 $this->state = self::STATE_CONNECTED;
+
+                $this->connectionMonitorWatcherId =  Loop::repeat(
+                    self::CONNECTION_MONITOR_INTERVAL,
+                    function(): void
+                    {
+                        if($this->connection->connected() === false) {
+                            throw Exception\ClientException::disconnected();
+                        }
+                    }
+                );
             }
         );
     }
@@ -153,6 +171,12 @@ function () use ($code, $reason) {
                     throw Exception\ClientException::notConnected();
                 }
 
+                if($this->connectionMonitorWatcherId !== null){
+                    Loop::cancel($this->connectionMonitorWatcherId);
+
+                    $this->connectionMonitorWatcherId = null;
+                }
+
                 $this->state = self::STATE_DISCONNECTING;
 
                 if ($code === 0) {

src/Connection.php

@@ -58,16 +58,15 @@ final class Connection
      */
     private $heartbeatWatcherId;
 
-    /**
-     * @var callable|null
-     */
-    private $connectionLost;
-
-    public function __construct(string $uri, ?callable $connectionLost = null)
+    public function __construct(string $uri)
     {
         $this->uri = $uri;
         $this->parser = new Parser;
-        $this->connectionLost = $connectionLost;
+    }
+
+    public function connected(): bool
+    {
+        return $this->socket !== null && $this->socket->isClosed() === false;
     }
 
     /**
@@ -195,12 +194,10 @@ function (string $watcherId) use ($interval){
                 }
 
                 if (
-                    null !== $this->connectionLost &&
                     0 !== $this->lastRead &&
                     $currentTime > ($this->lastRead + $interval + 1000)
                 )
                 {
-                    call_user_func($this->connectionLost);
                     Loop::cancel($watcherId);
                 }
 
@@ -218,10 +215,6 @@ public function close(): void
             $this->heartbeatWatcherId = null;
         }
 
-        if ($this->connectionLost !== null) {
-            call_user_func($this->connectionLost);
-        }
-
         if ($this->socket !== null) {
             $this->socket->close();
         }

src/Exception/ClientException.php

@@ -26,6 +26,10 @@ public static function notConnected(): self
         return new self('Client is not connected to server.');
     }
 
+    public static function disconnected(): self {
+        return new self('The client was unexpectedly disconnected from the server');
+    }
+
     public static function alreadyConnected(): self
     {
         return new self('Client is already connected/connecting.');