lib/CosmoW/Riak/Collection.php
protected function doAggregateCursor(array $pipeline, array $options = array())
    {
        if ( ! method_exists('RiakCollection', 'aggregateCursor')) {
            throw new BadMethodCallException('RiakCollection::aggregateCursor() is not available');
        }

        list($commandOptions, $clientOptions) = isset($options['socketTimeoutMS']) || isset($options['timeout'])
            ? $this->splitCommandAndClientOptions($options)
            : array($options, array());

        if (is_scalar($commandOptions['cursor'])) {
            unset($commandOptions['cursor']);
        }

        $timeout = isset($clientOptions['socketTimeoutMS'])
            ? $clientOptions['socketTimeoutMS']
            : (isset($clientOptions['timeout']) ? $clientOptions['timeout'] : null);

        $riakCollection = $this->riakCollection;
        $commandCursor = $this->retry(function() use ($riakCollection, $pipeline, $commandOptions) {
            return $riakCollection->aggregateCursor($pipeline, $commandOptions);
        });

        $commandCursor = $this->wrapCommandCursor($commandCursor);

        if (isset($timeout)) {
            $commandCursor->timeout($timeout);
        }

        return $commandCursor;
    }
protected function doAggregateCursor(array $pipeline, array $options = array())
    {
        if ( ! method_exists('RiakCollection', 'aggregateCursor')) {
            throw new BadMethodCallException('RiakCollection::aggregateCursor() is not available');
        }

        list($commandOptions, $clientOptions) = isset($options['socketTimeoutMS']) || isset($options['timeout'])
            ? $this->splitCommandAndClientOptions($options)
            : array($options, array());

        if (is_scalar($commandOptions['cursor'])) {
            unset($commandOptions['cursor']);
        }

        $timeout = isset($clientOptions['socketTimeoutMS'])
            ? $clientOptions['socketTimeoutMS']
            : (isset($clientOptions['timeout']) ? $clientOptions['timeout'] : null);

        $riakCollection = $this->riakCollection;
        $commandCursor = $this->retry(function() use ($riakCollection, $pipeline, $commandOptions) {
            return $riakCollection->aggregateCursor($pipeline, $commandOptions);
        });

        $commandCursor = $this->wrapCommandCursor($commandCursor);

        if (isset($timeout)) {
            $commandCursor->timeout($timeout);
        }

        return $commandCursor;
    }
protected function doAggregateCursor ( array $ pipeline , array $ options = array ( ) ) { if ( ! method_exists ( 'RiakCollection' , 'aggregateCursor' ) ) { throw new BadMethodCallException ( 'RiakCollection::aggregateCursor() is not available' ) ; } list ( $ commandOptions , $ clientOptions ) = isset ( $ options [ 'socketTimeoutMS' ] ) || isset ( $ options [ 'timeout' ] ) ? $ this -> splitCommandAndClientOptions ( $ options ) : array ( $ options , array ( ) ) ; if ( is_scalar ( $ commandOptions [ 'cursor' ] ) ) { unset ( $ commandOptions [ 'cursor' ] ) ; } $ timeout = isset ( $ clientOptions [ 'socketTimeoutMS' ] ) ? $ clientOptions [ 'socketTimeoutMS' ] : ( isset ( $ clientOptions [ 'timeout' ] ) ? $ clientOptions [ 'timeout' ] : null ) ; $ riakCollection = $ this -> riakCollection ; $ commandCursor = $ this -> retry ( function ( ) use ( $ riakCollection , $ pipeline , $ commandOptions ) { return $ riakCollection -> aggregateCursor ( $ pipeline , $ commandOptions ) ; } ) ; $ commandCursor = $ this -> wrapCommandCursor ( $ commandCursor ) ; if ( isset ( $ timeout ) ) { $ commandCursor -> timeout ( $ timeout ) ; } return $ commandCursor ; }
Executes the aggregate command and returns a RiakCommandCursor.

@param array $pipeline
@param array $options
@return CommandCursor
@throws BadMethodCallException if RiakCollection::aggregateCursor() is not available
Executes the aggregate command and returns a RiakCommandCursor .