Better Keep Alive support

Author: jjok

examples/config.php.example

@@ -3,7 +3,9 @@
 $config = array(
     'server' => 'yourMqttBroker.tld',
     'port' => 1883,
-    'options' => null,
+    'options' => new \oliverlorenz\reactphpmqtt\packet\ConnectionOptions(array(
+        'keepAlive' => 120,
+    )),
 );
 
 return $config;

src/Connector.php

@@ -76,8 +76,8 @@ public function create(
             ->then(function (Stream $stream) {
                 return $this->listenForPackets($stream);
             })
-            ->then(function(Stream $stream) {
-                return $this->keepAlive($stream);
+            ->then(function(Stream $stream) use ($options) {
+                return $this->keepAlive($stream, $options->keepAlive);
             });
     }
 
@@ -119,12 +119,16 @@ private function listenForPackets(Stream $stream)
         return $deferred->promise();
     }
 
-    private function keepAlive(Stream $stream)
+    private function keepAlive(Stream $stream, $keepAlive)
     {
-        $this->getLoop()->addPeriodicTimer(10, function(Timer $timer) use ($stream) {
-            $packet = new PingRequest($this->version);
-            $this->sendPacketToStream($stream, $packet);
-        });
+        if($keepAlive > 0) {
+            $interval = (int) ($keepAlive / 2);
+
+            $this->getLoop()->addPeriodicTimer($interval, function(Timer $timer) use ($stream) {
+                $packet = new PingRequest($this->version);
+                $this->sendPacketToStream($stream, $packet);
+            });
+        }
 
         return new FulfilledPromise($stream);
     }
@@ -142,7 +146,8 @@ public function connect(Stream $stream, ConnectionOptions $options) {
             $options->willTopic,
             $options->willMessage,
             $options->willQos,
-            $options->willRetain
+            $options->willRetain,
+            $options->keepAlive
         );
         $message = $packet->get();
         echo MessageHelper::getReadableByRawString($message);

src/packet/Connect.php

@@ -39,6 +39,9 @@ class Connect extends ControlPacket {
     /** @var null */
     protected $willRetain;
 
+    /** @var int */
+    private $keepAlive;
+
     /**
      * @param Version $version
      * @param string|null $username
@@ -49,6 +52,7 @@ class Connect extends ControlPacket {
      * @param string|null $willMessage
      * @param bool|null $willQos
      * @param null $willRetain
+     * @param int $keepAlive
      */
     public function __construct(
         Version $version,
@@ -59,7 +63,9 @@ public function __construct(
         $willTopic = null,
         $willMessage = null,
         $willQos = null,
-        $willRetain = null
+        $willRetain = null,
+//        $keepAlive = 0
+        $keepAlive = 10
     ) {
         parent::__construct($version);
         $this->clientId = $clientId;
@@ -70,6 +76,7 @@ public function __construct(
         $this->willMessage = $willMessage;
         $this->willQos = boolval($willQos);
         $this->willRetain = $willRetain;
+        $this->keepAlive = $keepAlive;
         $this->buildPayload();
     }
 
@@ -101,14 +108,24 @@ public static function getControlPacketType()
      */
     protected function getVariableHeader()
     {
-        return chr(ControlPacketType::MOST_SIGNIFICANT_BYTE)         // byte 1
-        . chr(strlen($this->version->getProtocolIdentifierString())) // byte 2
-        . $this->version->getProtocolIdentifierString()              // byte 3,4,5,6
-        . chr($this->version->getProtocolVersion())                  // byte 7
-        . chr($this->getConnectFlags())                              // byte 8
-        . chr(0)                                                     // byte 9
-        . chr(10)                                                    // byte 10
-        ;
+        return chr(ControlPacketType::MOST_SIGNIFICANT_BYTE)              // byte 1
+             . chr(strlen($this->version->getProtocolIdentifierString())) // byte 2
+             . $this->version->getProtocolIdentifierString()              // byte 3,4,5,6
+             . chr($this->version->getProtocolVersion())                  // byte 7
+             . chr($this->getConnectFlags())                              // byte 8
+             . $this->getKeepAlive();                                     // byte 9,10
+    }
+
+    /**
+     * @return string
+     */
+    private function getKeepAlive()
+    {
+        $msb = $this->keepAlive >> 8;
+        $lsb = $this->keepAlive % 256;
+
+        return chr($msb)
+             . chr($lsb);
     }
 
     /**

src/packet/ConnectionOptions.php

@@ -100,6 +100,16 @@ class ConnectionOptions
      */
     public $willRetain = false;
 
+    /**
+     * The Keep Alive is a time interval measured in seconds.
+     *
+     * @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Keep_Alive
+     *
+     * @var int
+     */
+//    public $keepAlive = 0;
+    public $keepAlive = 10;
+
     /**
      * ConnectionOptions constructor.
      *