lib/CosmoW/Riak/Collection.php
protected function doAggregate(array $pipeline, array $options = array())
    {
        if (isset($options['cursor']) && ($options['cursor'] || is_array($options['cursor']))) {
            return $this->doAggregateCursor($pipeline, $options);
        }

        unset($options['cursor']);

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

        $command = array();
        $command['aggregate'] = $this->riakCollection->getName();
        $command['pipeline'] = $pipeline;
        $command = array_merge($command, $commandOptions);

        $database = $this->database;
        $result = $this->retry(function() use ($database, $command, $clientOptions) {
            return $database->command($command, $clientOptions);
        });

        if (empty($result['ok'])) {
            throw new ResultException($result);
        }

        /* If the pipeline ends with an $out operator, return a cursor on that
         * collection so a table scan may be performed.
         */
        if (isset($pipeline[count($pipeline) - 1]['$out'])) {
            $outputCollection = $pipeline[count($pipeline) - 1]['$out'];

            return $database->selectCollection($outputCollection)->find();
        }

        $arrayIterator = new ArrayIterator(isset($result['result']) ? $result['result'] : array());
        $arrayIterator->setCommandResult($result);

        return $arrayIterator;
    }
protected function doAggregate(array $pipeline, array $options = array())
    {
        if (isset($options['cursor']) && ($options['cursor'] || is_array($options['cursor']))) {
            return $this->doAggregateCursor($pipeline, $options);
        }

        unset($options['cursor']);

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

        $command = array();
        $command['aggregate'] = $this->riakCollection->getName();
        $command['pipeline'] = $pipeline;
        $command = array_merge($command, $commandOptions);

        $database = $this->database;
        $result = $this->retry(function() use ($database, $command, $clientOptions) {
            return $database->command($command, $clientOptions);
        });

        if (empty($result['ok'])) {
            throw new ResultException($result);
        }

        /* If the pipeline ends with an $out operator, return a cursor on that
         * collection so a table scan may be performed.
         */
        if (isset($pipeline[count($pipeline) - 1]['$out'])) {
            $outputCollection = $pipeline[count($pipeline) - 1]['$out'];

            return $database->selectCollection($outputCollection)->find();
        }

        $arrayIterator = new ArrayIterator(isset($result['result']) ? $result['result'] : array());
        $arrayIterator->setCommandResult($result);

        return $arrayIterator;
    }
protected function doAggregate ( array $ pipeline , array $ options = array ( ) ) { if ( isset ( $ options [ 'cursor' ] ) && ( $ options [ 'cursor' ] || is_array ( $ options [ 'cursor' ] ) ) ) { return $ this -> doAggregateCursor ( $ pipeline , $ options ) ; } unset ( $ options [ 'cursor' ] ) ; list ( $ commandOptions , $ clientOptions ) = isset ( $ options [ 'socketTimeoutMS' ] ) || isset ( $ options [ 'timeout' ] ) ? $ this -> splitCommandAndClientOptions ( $ options ) : array ( $ options , array ( ) ) ; $ command = array ( ) ; $ command [ 'aggregate' ] = $ this -> riakCollection -> getName ( ) ; $ command [ 'pipeline' ] = $ pipeline ; $ command = array_merge ( $ command , $ commandOptions ) ; $ database = $ this -> database ; $ result = $ this -> retry ( function ( ) use ( $ database , $ command , $ clientOptions ) { return $ database -> command ( $ command , $ clientOptions ) ; } ) ; if ( empty ( $ result [ 'ok' ] ) ) { throw new ResultException ( $ result ) ; } /* If the pipeline ends with an $out operator, return a cursor on that
         * collection so a table scan may be performed.
         */ if ( isset ( $ pipeline [ count ( $ pipeline ) - 1 ] [ '$out' ] ) ) { $ outputCollection = $ pipeline [ count ( $ pipeline ) - 1 ] [ '$out' ] ; return $ database -> selectCollection ( $ outputCollection ) -> find ( ) ; } $ arrayIterator = new ArrayIterator ( isset ( $ result [ 'result' ] ) ? $ result [ 'result' ] : array ( ) ) ; $ arrayIterator -> setCommandResult ( $ result ) ; return $ arrayIterator ; }
Execute the aggregate command.

@see Collection::aggregate()
@param array $pipeline
@param array $options
@return Iterator
Execute the aggregate command .