private function timeout(Observable $events): Observable
    {
        $timeoutDuration = $this->connected->map(function (Event $event) {
            return ($event->getData()['activity_timeout'] ?? self::NO_ACTIVITY_TIMEOUT) * 1000;
        });

        return $timeoutDuration
            ->combineLatest([$events])
            ->pluck(0)
            ->concat(Observable::of(-1))
            ->flatMapLatest(function (int $time) {

                // If the events observable ends, return an empty observable so we don't keep the stream alive
                if ($time === -1) {
                    return Observable::empty();
                }

                return Observable::never()
                    ->timeout($time)
                    ->catch(function () use ($time) {
                        // ping (do something that causes incoming stream to get a message)
                        $this->send(Event::ping());
                        // this timeout will actually timeout with a TimeoutException - causing
                        // everything above this to dispose
                        return Observable::never()->timeout($time);
                    });
            });
    }