public function aggregate(array $pipeline, array $options = [])
    {
        $hasOutStage = \MongoDB\is_last_pipeline_operator_out($pipeline);

        if ( ! isset($options['readPreference'])) {
            $options['readPreference'] = $this->readPreference;
        }

        if ($hasOutStage) {
            $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
        }

        $server = $this->manager->selectServer($options['readPreference']);

        /* A "majority" read concern is not compatible with the $out stage, so
         * avoid providing the Collection's read concern if it would conflict.
         *
         * A read concern is also not compatible with transactions.
         */
        if ( ! isset($options['readConcern']) &&
            ! ($hasOutStage && $this->readConcern->getLevel() === ReadConcern::MAJORITY) &&
            \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern) &&
            ! $this->isInTransaction($options)) {
            $options['readConcern'] = $this->readConcern;
        }

        if ( ! isset($options['typeMap'])) {
            $options['typeMap'] = $this->typeMap;
        }

        if ($hasOutStage &&
            ! isset($options['writeConcern']) &&
            \MongoDB\server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) &&
            ! $this->isInTransaction($options)) {
            $options['writeConcern'] = $this->writeConcern;
        }

        $operation = new Aggregate($this->databaseName, $this->collectionName, $pipeline, $options);

        return $operation->execute($server);
    }
public function aggregate(array $pipeline, array $options = [])
    {
        $hasOutStage = \MongoDB\is_last_pipeline_operator_out($pipeline);

        if ( ! isset($options['readPreference'])) {
            $options['readPreference'] = $this->readPreference;
        }

        if ($hasOutStage) {
            $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
        }

        $server = $this->manager->selectServer($options['readPreference']);

        /* A "majority" read concern is not compatible with the $out stage, so
         * avoid providing the Collection's read concern if it would conflict.
         *
         * A read concern is also not compatible with transactions.
         */
        if ( ! isset($options['readConcern']) &&
            ! ($hasOutStage && $this->readConcern->getLevel() === ReadConcern::MAJORITY) &&
            \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern) &&
            ! $this->isInTransaction($options)) {
            $options['readConcern'] = $this->readConcern;
        }

        if ( ! isset($options['typeMap'])) {
            $options['typeMap'] = $this->typeMap;
        }

        if ($hasOutStage &&
            ! isset($options['writeConcern']) &&
            \MongoDB\server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) &&
            ! $this->isInTransaction($options)) {
            $options['writeConcern'] = $this->writeConcern;
        }

        $operation = new Aggregate($this->databaseName, $this->collectionName, $pipeline, $options);

        return $operation->execute($server);
    }
public function aggregate ( array $ pipeline , array $ options = [ ] ) { $ hasOutStage = \ MongoDB \ is_last_pipeline_operator_out ( $ pipeline ) ; if ( ! isset ( $ options [ 'readPreference' ] ) ) { $ options [ 'readPreference' ] = $ this -> readPreference ; } if ( $ hasOutStage ) { $ options [ 'readPreference' ] = new ReadPreference ( ReadPreference :: RP_PRIMARY ) ; } $ server = $ this -> manager -> selectServer ( $ options [ 'readPreference' ] ) ; /* A "majority" read concern is not compatible with the $out stage, so
         * avoid providing the Collection's read concern if it would conflict.
         *
         * A read concern is also not compatible with transactions.
         */ if ( ! isset ( $ options [ 'readConcern' ] ) && ! ( $ hasOutStage && $ this -> readConcern -> getLevel ( ) === ReadConcern :: MAJORITY ) && \ MongoDB \ server_supports_feature ( $ server , self :: $ wireVersionForReadConcern ) && ! $ this -> isInTransaction ( $ options ) ) { $ options [ 'readConcern' ] = $ this -> readConcern ; } if ( ! isset ( $ options [ 'typeMap' ] ) ) { $ options [ 'typeMap' ] = $ this -> typeMap ; } if ( $ hasOutStage && ! isset ( $ options [ 'writeConcern' ] ) && \ MongoDB \ server_supports_feature ( $ server , self :: $ wireVersionForWritableCommandWriteConcern ) && ! $ this -> isInTransaction ( $ options ) ) { $ options [ 'writeConcern' ] = $ this -> writeConcern ; } $ operation = new Aggregate ( $ this -> databaseName , $ this -> collectionName , $ pipeline , $ options ) ; return $ operation -> execute ( $ server ) ; }
Executes an aggregation framework pipeline on the collection.

Note: this method's return value depends on the MongoDB server version
and the "useCursor" option. If "useCursor" is true, a Cursor will be
returned; otherwise, an ArrayIterator is returned, which wraps the
"result" array from the command response document.

@see Aggregate::__construct() for supported options
@param array $pipeline List of pipeline operations
@param array $options  Command options
@return Traversable
@throws UnexpectedValueException if the command response was malformed
@throws UnsupportedException if options are not supported by the selected server
@throws InvalidArgumentException for parameter/option parsing errors
@throws DriverRuntimeException for other driver errors (e.g. connection errors)
Executes an aggregation framework pipeline on the collection .