Send only one packet per tick

Author: binsoul

src/ReactMqttClient.php

@@ -72,6 +72,8 @@ class ReactMqttClient extends EventEmitter
     private $publishQos2 = [];
     /** @var PublishRequestPacket[] */
     private $incomingQos2 = [];
+    /** @var Packet[] */
+    private $packetQueue = [];
 
     /**
      * Constructs an instance of this class.
@@ -171,8 +173,7 @@ public function disconnect()
 
         $packet = new DisconnectRequestPacket();
 
-        $this->stream->write($packet);
-        $this->stream->close();
+        $this->stream->end($packet);
         $this->stream = null;
 
         $this->isConnected = false;
@@ -192,7 +193,7 @@ public function subscribe($topic, $qosLevel = 0)
         $packet->setTopic($topic);
         $packet->setQosLevel($qosLevel);
 
-        $this->stream->write($packet);
+        $this->writePacket($packet);
 
         $id = $packet->getIdentifier();
         $this->deferred['subscribe'][$id] = new Deferred();
@@ -213,7 +214,7 @@ public function unsubscribe($topic)
         $packet = new UnsubscribeRequestPacket();
         $packet->setTopic($topic);
 
-        $this->stream->write($packet);
+        $this->writePacket($packet);
 
         $id = $packet->getIdentifier();
         $this->deferred['unsubscribe'][$id] = new Deferred();
@@ -241,7 +242,7 @@ public function publish($topic, $message, $qosLevel = 0, $retain = false)
         $packet->setRetained($retain);
         $packet->setDuplicate(false);
 
-        $this->stream->write($packet);
+        $this->writePacket($packet);
 
         if ($qosLevel == 0) {
             return new FulfilledPromise($message);
@@ -305,6 +306,13 @@ private function connectStream(Stream $stream, array $settings, $timeout, $keepA
             $this->handleData($data);
         });
 
+        $this->stream->getBuffer()->on('full-drain', function () {
+            if (count($this->packetQueue) > 0) {
+                $packet = array_shift($this->packetQueue);
+                $this->stream->write($packet);
+            }
+        });
+
         $this->stream->on('error', function (\Exception $e) {
             $this->emitError($e);
         });
@@ -321,7 +329,7 @@ private function connectStream(Stream $stream, array $settings, $timeout, $keepA
         $this->timer[] = $this->loop->addPeriodicTimer(
             floor($keepAlive * 0.75),
             function () {
-                $this->stream->write(new PingRequestPacket());
+                $this->writePacket(new PingRequestPacket());
             }
         );
 
@@ -349,7 +357,7 @@ function () use ($timeout) {
             $packet->setWill($will['topic'], $will['message'], $will['qos'], $will['retain']);
         }
 
-        $this->stream->write($packet);
+        $this->writePacket($packet);
     }
 
     /**
@@ -460,6 +468,20 @@ private function sanitizeOptions(array $options)
         return array_merge($defaults, $options);
     }
 
+    /**
+     * Writes a packet to the stream or buffers it if the stream is busy.
+     *
+     * @param Packet $packet
+     */
+    private function writePacket(Packet $packet)
+    {
+        if ($this->stream->getBuffer()->listening) {
+            $this->packetQueue[] = $packet;
+        } else {
+            $this->stream->write($packet);
+        }
+    }
+
     /**
      * Handles a CONNACK packet.
      *
@@ -563,7 +585,7 @@ private function handlePublishRequest(PublishRequestPacket $packet)
 
         if ($response !== null) {
             $response->setIdentifier($packet->getIdentifier());
-            $this->stream->write($response);
+            $this->writePacket($response);
         }
 
         if ($emit) {
@@ -606,7 +628,7 @@ private function handlePublishReceived(PublishReceivedPacket $packet)
 
         $response = new PublishReleasePacket();
         $response->setIdentifier($id);
-        $this->stream->write($response);
+        $this->writePacket($response);
     }
 
     /**
@@ -620,7 +642,7 @@ private function handlePublishRelease(PublishReleasePacket $packet)
 
         $response = new PublishCompletePacket();
         $response->setIdentifier($id);
-        $this->stream->write($response);
+        $this->writePacket($response);
 
         if (!isset($this->incomingQos2[$id])) {
             $this->emitWarning(new \LogicException(sprintf('PUBREL: Packet identifier %d not found.', $id)));