src/AsyncClient.php
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);
                    });
            });
    }
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);
                    });
            });
    }
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 ) ; } ) ; } ) ; }
Returns an observable of TimeoutException.
The timeout observable will get cancelled every time a new event is received.

@param  Observable $events
@return Observable
Returns an observable of TimeoutException . The timeout observable will get cancelled every time a new event is received .