Make connect and disconnect flows silent

Author: binsoul

src/ReactFlow.php

@@ -17,19 +17,23 @@ class ReactFlow implements Flow
     private $deferred;
     /** @var Packet */
     private $packet;
+    /** @var bool */
+    private $isSilent;
 
     /**
      * Constructs an instance of this class.
      *
      * @param Flow     $decorated
      * @param Deferred $deferred
      * @param Packet   $packet
+     * @param bool     $isSilent
      */
-    public function __construct(Flow $decorated, Deferred $deferred, Packet $packet = null)
+    public function __construct(Flow $decorated, Deferred $deferred, Packet $packet = null, $isSilent = false)
     {
         $this->decorated = $decorated;
         $this->deferred = $deferred;
         $this->packet = $packet;
+        $this->isSilent = $isSilent;
     }
 
     public function getCode()
@@ -95,4 +99,14 @@ public function getPacket()
     {
         return $this->packet;
     }
+
+    /**
+     * Indicates if the flow should emit events.
+     *
+     * @return bool
+     */
+    public function isSilent()
+    {
+        return $this->isSilent;
+    }
 }

src/ReactMqttClient.php

@@ -198,6 +198,7 @@ public function connect($host, $port = 1883, Connection $connection = null, $tim
                         $this->isConnected = true;
                         $this->connection = $connection;
 
+                        $this->emit('connect', [$connection, $this]);
                         $deferred->resolve($this->connection);
                     })
                     ->otherwise(function (\Exception $e) use ($deferred, $connection) {
@@ -235,14 +236,15 @@ public function disconnect()
 
         $deferred = new Deferred();
 
-        $this->startFlow(new OutgoingDisconnectFlow($this->connection))
+        $this->startFlow(new OutgoingDisconnectFlow($this->connection), true)
             ->then(function (Connection $connection) use ($deferred) {
                 $this->isDisconnecting = false;
                 $this->isConnected = false;
 
-                $this->stream->close();
-
+                $this->emit('disconnect', [$connection, $this]);
                 $deferred->resolve($connection);
+
+                $this->stream->close();
             })
             ->otherwise(function () use ($deferred) {
                 $this->isDisconnecting = false;
@@ -425,7 +427,7 @@ function () use ($deferred, $timeout) {
             }
         );
 
-        $this->startFlow(new OutgoingConnectFlow($connection, $this->identifierGenerator))
+        $this->startFlow(new OutgoingConnectFlow($connection, $this->identifierGenerator), true)
             ->always(function () use ($responseTimer) {
                 $this->loop->cancelTimer($responseTimer);
             })->then(function (Connection $connection) use ($deferred) {
@@ -583,10 +585,11 @@ private function handleError(\Exception $e)
      * Starts the given flow.
      *
      * @param Flow $flow
+     * @param bool $isSilent
      *
      * @return ExtendedPromiseInterface
      */
-    private function startFlow(Flow $flow)
+    private function startFlow(Flow $flow, $isSilent = false)
     {
         try {
             $packet = $flow->start();
@@ -597,7 +600,7 @@ private function startFlow(Flow $flow)
         }
 
         $deferred = new Deferred();
-        $internalFlow = new ReactFlow($flow, $deferred, $packet);
+        $internalFlow = new ReactFlow($flow, $deferred, $packet, $isSilent);
 
         if ($packet !== null) {
             if ($this->stream->getBuffer()->listening) {
@@ -653,7 +656,9 @@ private function continueFlow(ReactFlow $flow, Packet $packet)
     private function finishFlow(ReactFlow $flow)
     {
         if ($flow->isSuccess()) {
-            $this->emit($flow->getCode(), [$flow->getResult(), $this]);
+            if (!$flow->isSilent()) {
+                $this->emit($flow->getCode(), [$flow->getResult(), $this]);
+            }
 
             $flow->getDeferred()->resolve($flow->getResult());
         } else {

tests/Integration/ReactMqttClientTest.php

@@ -103,6 +103,8 @@ public function setUp()
             $this->loop->stop();
             $this->fail('Test timeout');
         });
+
+        echo 'Test: '.str_replace(['test_', '_'], ['', ' '], $this->getName()).PHP_EOL.PHP_EOL;
     }
 
     /**
@@ -112,7 +114,7 @@ public function tearDown()
     {
         $this->loop->stop();
 
-        $this->log(str_repeat('- - ', 25));
+        echo str_repeat('- - ', 25).PHP_EOL;
     }
 
     /*******************************************************
@@ -335,25 +337,51 @@ public function test_connect_failure()
     }
 
     /**
-     * Test that client's is-connected state is updated correctly
+     * Test that client's connection state is updated correctly when connected.
      *
      * @depends test_connect_success
      */
     public function test_is_connected_when_connect_event_emitted()
     {
         $client = $this->buildClient();
 
-        $client->on('connect', function(Connection $connection) use($client){
+        $client->on('connect', function () use ($client) {
             $this->assertTrue($client->isConnected(), 'Client is should be connected');
             $this->stopLoop();
         });
 
-        $client->connect(self::HOSTNAME, self::PORT, null, 1)
-        ->then(function () use ($client) {
-            $this->assertTrue($client->isConnected());
+        $client->connect(self::HOSTNAME, self::PORT)
+            ->then(function () use ($client) {
+                $this->assertTrue($client->isConnected());
+                $this->stopLoop();
+            });
+
+        $this->startLoop();
+    }
+
+    /**
+     * Test that client's connection state is updated correctly when disconnected.
+     *
+     * @depends test_connect_success
+     */
+    public function test_is_disconnected_when_disconnect_event_emitted()
+    {
+        $client = $this->buildClient();
+
+        $client->on('disconnect', function () use ($client) {
+            $this->assertFalse($client->isConnected(), 'Client is should be disconnected');
             $this->stopLoop();
         });
 
+        $client->connect(self::HOSTNAME, self::PORT)
+            ->then(function () use ($client) {
+                $client->disconnect()
+                    ->then(function () use ($client) {
+                        $this->assertFalse($client->isConnected());
+                        $this->stopLoop();
+                    });
+            });
+
         $this->startLoop();
     }