function collect() {
        return (new \React\Promise\Promise(function (callable $resolve, callable $reject) {
            $this->resolve = $resolve;
            $this->reject = $reject;
            
            $filter = $this->filter;
            $handler = $this->handler;
            
            $this->listener = function (...$item) use (&$filter, &$handler) {
                if($filter === null || $filter(...$item)) {
                    list($key, $value) = $handler(...$item);
                    $this->bucket->set($key, $value);
                    
                    if(($this->options['max'] ?? \INF) <= $this->bucket->count()) {
                        $this->stop();
                    }
                }
            };
            
            if(($this->options['time'] ?? 30) > 0) {
                $this->timer = self::$loop->addTimer(($this->options['time'] ?? 30), function () {
                    $this->stop();
                });
            }
            
            $this->emitter->on($this->event, $this->listener);
        }, function (callable $resolve, callable $reject) {
            if($this->timer) {
                self::$loop->cancelTimer($this->timer);
                $this->timer = null;
            }
            
            $this->emitter->removeListener($this->event, $this->listener);
            $reject(new \OutOfBoundsException('Operation cancelled'));
        }));
    }
function collect() {
        return (new \React\Promise\Promise(function (callable $resolve, callable $reject) {
            $this->resolve = $resolve;
            $this->reject = $reject;
            
            $filter = $this->filter;
            $handler = $this->handler;
            
            $this->listener = function (...$item) use (&$filter, &$handler) {
                if($filter === null || $filter(...$item)) {
                    list($key, $value) = $handler(...$item);
                    $this->bucket->set($key, $value);
                    
                    if(($this->options['max'] ?? \INF) <= $this->bucket->count()) {
                        $this->stop();
                    }
                }
            };
            
            if(($this->options['time'] ?? 30) > 0) {
                $this->timer = self::$loop->addTimer(($this->options['time'] ?? 30), function () {
                    $this->stop();
                });
            }
            
            $this->emitter->on($this->event, $this->listener);
        }, function (callable $resolve, callable $reject) {
            if($this->timer) {
                self::$loop->cancelTimer($this->timer);
                $this->timer = null;
            }
            
            $this->emitter->removeListener($this->event, $this->listener);
            $reject(new \OutOfBoundsException('Operation cancelled'));
        }));
    }
function collect ( ) { return ( new \ React \ Promise \ Promise ( function ( callable $ resolve , callable $ reject ) { $ this -> resolve = $ resolve ; $ this -> reject = $ reject ; $ filter = $ this -> filter ; $ handler = $ this -> handler ; $ this -> listener = function ( ... $ item ) use ( & $ filter , & $ handler ) { if ( $ filter === null || $ filter ( ... $ item ) ) { list ( $ key , $ value ) = $ handler ( ... $ item ) ; $ this -> bucket -> set ( $ key , $ value ) ; if ( ( $ this -> options [ 'max' ] ?? \ INF ) <= $ this -> bucket -> count ( ) ) { $ this -> stop ( ) ; } } } ; if ( ( $ this -> options [ 'time' ] ?? 30 ) > 0 ) { $ this -> timer = self :: $ loop -> addTimer ( ( $ this -> options [ 'time' ] ?? 30 ) , function ( ) { $ this -> stop ( ) ; } ) ; } $ this -> emitter -> on ( $ this -> event , $ this -> listener ) ; } , function ( callable $ resolve , callable $ reject ) { if ( $ this -> timer ) { self :: $ loop -> cancelTimer ( $ this -> timer ) ; $ this -> timer = null ; } $ this -> emitter -> removeListener ( $ this -> event , $ this -> listener ) ; $ reject ( new \ OutOfBoundsException ( 'Operation cancelled' ) ) ; } ) ) ; }
Starts collecting. Resolves with a collection.
@return \React\Promise\ExtendedPromiseInterface   This promise is cancellable.
@throws \RangeException          The exception the promise gets rejected with, if collecting times out.
@throws \OutOfBoundsException    The exception the promise gets rejected with, if the promise gets cancelled.
Starts collecting . Resolves with a collection .