Merge pull request #33 from PNixx/master

fix heartbeat interval

Author: mmasiukevich

src/Client.php

@@ -91,7 +91,7 @@ function () {
 
                 $this->state = self::STATE_CONNECTING;
 
-                $this->connection = new Connection($this->config->uri());
+                $this->connection = new Connection($this->config->uri(), fn() => $this->state = self::STATE_NOT_CONNECTED);
 
                 yield $this->connection->open(
                     $this->config->timeout,
@@ -287,7 +287,7 @@ function () {
                 $heartbeatInterval = $this->config->heartbeat;
 
                 if ($heartbeatInterval !== 0) {
-                    $heartbeatInterval = \min($heartbeatInterval, $tune->heartbeat);
+                    $heartbeatInterval = \min($heartbeatInterval, $tune->heartbeat * 1000);
                 }
 
                 $maxChannel = \min($this->config->maxChannel, $tune->channelMax);
@@ -302,17 +302,15 @@ function () {
                     ->appendUint16(31)
                     ->appendInt16($maxChannel)
                     ->appendInt32($maxFrame)
-                    ->appendInt16($heartbeatInterval)
+                    ->appendInt16((int) ($heartbeatInterval / 1000))
                     ->appendUint8(206);
 
                 yield $this->connection->write($buffer);
 
                 $this->properties->tune($maxChannel, $maxFrame);
 
                 if ($heartbeatInterval > 0) {
-                    $this->connection->heartbeat($heartbeatInterval, function (){
-                        $this->state = self::STATE_NOT_CONNECTED;
-                    });
+                    $this->connection->heartbeat($heartbeatInterval);
                 }
             }
         );

src/Connection.php

@@ -58,10 +58,16 @@ final class Connection
      */
     private $heartbeatWatcherId;
 
-    public function __construct(string $uri)
+    /**
+     * @var callable|null
+     */
+    private $connectionLost;
+
+    public function __construct(string $uri, ?callable $connectionLost = null)
     {
         $this->uri = $uri;
         $this->parser = new Parser;
+        $this->connectionLost = $connectionLost;
     }
 
     /**
@@ -164,11 +170,11 @@ function () {
         );
     }
 
-    public function heartbeat(int $interval, ?callable $connectionLost = null): void
+    public function heartbeat(int $interval): void
     {
         $this->heartbeatWatcherId = Loop::repeat(
             $interval,
-            function (string $watcherId) use ($interval, $connectionLost){
+            function (string $watcherId) use ($interval){
                 $currentTime = Loop::now();
 
                 if (null !== $this->socket) {
@@ -189,12 +195,12 @@ function (string $watcherId) use ($interval, $connectionLost){
                 }
 
                 if (
-                    null !== $connectionLost &&
+                    null !== $this->connectionLost &&
                     0 !== $this->lastRead &&
                     $currentTime > ($this->lastRead + $interval + 1000)
                 )
                 {
-                    $connectionLost();
+                    call_user_func($this->connectionLost);
                     Loop::cancel($watcherId);
                 }
 
@@ -212,6 +218,10 @@ public function close(): void
             $this->heartbeatWatcherId = null;
         }
 
+        if ($this->connectionLost !== null) {
+            call_user_func($this->connectionLost);
+        }
+
         if ($this->socket !== null) {
             $this->socket->close();
         }