Resolve connect and disconnect with the result of the flow

Author: binsoul

src/ReactMqttClient.php

@@ -202,15 +202,15 @@ public function connect($host, $port = 1883, Connection $connection = null, $tim
                 $this->emit('open', [$connection, $this]);
 
                 $this->registerClient($connection, $timeout)
-                    ->then(function (Connection $connection) use ($deferred) {
+                    ->then(function ($result) use ($connection, $deferred) {
                         $this->isConnecting = false;
                         $this->isConnected = true;
                         $this->connection = $connection;
 
                         $this->emit('connect', [$connection, $this]);
-                        $deferred->resolve($this->connection);
+                        $deferred->resolve($result ?: $connection);
                     })
-                    ->otherwise(function (\Exception $e) use ($deferred, $connection) {
+                    ->otherwise(function (\Exception $e) use ($connection, $deferred) {
                         $this->isConnecting = false;
 
                         $this->emitError($e);
@@ -251,20 +251,24 @@ public function disconnect($timeout = 5)
         $deferred = new Deferred();
 
         $isResolved = false;
+        $flowResult = null;
 
-        $this->onCloseCallback = function ($connection) use ($deferred, &$isResolved) {
+        $this->onCloseCallback = function ($connection) use ($deferred, &$isResolved, &$flowResult) {
             if (!$isResolved) {
                 $isResolved = true;
 
                 if ($connection) {
                     $this->emit('disconnect', [$connection, $this]);
-                    $deferred->resolve($connection);
                 }
+
+                $deferred->resolve($flowResult ?: $connection);
             }
         };
 
         $this->startFlow($this->flowFactory->buildOutgoingDisconnectFlow($this->connection), true)
-            ->then(function () use ($timeout) {
+            ->then(function ($result) use ($timeout, &$flowResult) {
+                $flowResult = $result;
+
                 $this->timer[] = $this->loop->addTimer(
                     $timeout,
                     function () {
@@ -274,11 +278,11 @@ function () {
                     }
                 );
             })
-            ->otherwise(function () use ($deferred, &$isResolved) {
+            ->otherwise(function ($exception) use ($deferred, &$isResolved) {
                 if (!$isResolved) {
                     $isResolved = true;
                     $this->isDisconnecting = false;
-                    $deferred->reject($this->connection);
+                    $deferred->reject($exception);
                 }
             });
 
@@ -461,15 +465,15 @@ function () use ($deferred, $timeout) {
         $this->startFlow($this->flowFactory->buildOutgoingConnectFlow($connection), true)
             ->always(function () use ($responseTimer) {
                 $this->loop->cancelTimer($responseTimer);
-            })->then(function (Connection $connection) use ($deferred) {
+            })->then(function ($result) use ($connection, $deferred) {
                 $this->timer[] = $this->loop->addPeriodicTimer(
                     floor($connection->getKeepAlive() * 0.75),
                     function () {
                         $this->startFlow($this->flowFactory->buildOutgoingPingFlow());
                     }
                 );
 
-                $deferred->resolve($connection);
+                $deferred->resolve($result ?: $connection);
             })->otherwise(function (\Exception $e) use ($deferred) {
                 $deferred->reject($e);
             });