Skip to content

Commit

Permalink
Avoid closing already closed streams
Browse files Browse the repository at this point in the history
  • Loading branch information
@binsoul
binsoul committed May 17, 2017
1 parent 43779e3 commit 091bee2
Showing 1 changed file with 7 additions and 2 deletions.
9 changes: 7 additions & 2 deletions src/ReactMqttClient.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -207,7 +207,10 @@ public function connect($host, $port = 1883, Connection $connection = null, $tim
$this->emitError($e);
$deferred->reject($e);

$this->stream->close();
if ($this->stream !== null) {
$this->stream->close();
}

$this->emit('close', [$connection, $this]);
});
})
Expand Down Expand Up @@ -244,7 +247,9 @@ public function disconnect()
$this->emit('disconnect', [$connection, $this]);
$deferred->resolve($connection);

$this->stream->close();
if ($this->stream !== null) {

This comment has been minimized.

Sign in to view

Copy link
@brstgt

brstgt Oct 17, 2017

There is a race condition between active and passive connection closing. In my case this ends in E_WARNINGS from ext/Event. Normally the server closes the connection after a successful disconnect message. So the active connection close should not be necessary and the client should be informed by the stream about the closed connection. However, if sth goes wrong, there should be a fallback.

I suggest this

        $deferred = new Deferred();

        $resolved = false;
        $finishDisconnect = function($connection) use ($deferred, &$resolved) {
            if ($resolved) {
                return;
            }
            $resolved = true;
            $this->isDisconnecting = false;
            $this->emit('disconnect', [$connection, $this]);
            $deferred->resolve($connection);
        };

        $timer = null;
        $this->once('close', function($connection) use ($finishDisconnect, &$timer) {
            if ($timer) {
                $this->loop->cancelTimer($timer);
            }
            $finishDisconnect($connection);
        });

        $this->startFlow(new OutgoingDisconnectFlow($this->connection), true)
            ->then(function (Connection $connection) use ($finishDisconnect, &$timer) {
                $this->isConnected = false;

                $timer = $this->loop->addTimer(0.1, function () use ($connection, $finishDisconnect, &$timer) {
                    $timer = null;
                    if ($this->stream !== null) {
                        $finishDisconnect($connection);
                        $this->stream->close();
                    }
                });
            })
            ->otherwise(function () use ($deferred, &$resolved) {
                $this->isDisconnecting = false;
                if ($resolved) {
                    return;
                }
                $resolved = true;
                $deferred->reject($this->connection);
            });
$this->stream->close();
}
})
->otherwise(function () use ($deferred) {
$this->isDisconnecting = false;
Expand Down

0 comments on commit 091bee2

Please sign in to comment.