PHPLIB-86: Fix aggregate() useCursor default and 2.4 compatibility

This adds intelligent detection for the useCursor option's default value. Additionally, server 2.4 does not support any of the aggregate command options of newer servers, so we must not specify default values.

The array_map() conversion for inline aggregation results is not ideal, but we can revisit that in PHPLIB-100.

Author: jmikola

src/Collection.php

@@ -71,37 +71,75 @@ public function __construct(Manager $manager, $namespace, WriteConcern $writeCon
 
     /**
      * Runs an aggregation framework pipeline
-     * NOTE: The return value of this method depends on your MongoDB server version
-     *       and possibly options.
-     *       MongoDB 2.6 (and later) will return a Cursor by default
-     *       MongoDB pre 2.6 will return an ArrayIterator
+     *
+     * 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 http://docs.mongodb.org/manual/reference/command/aggregate/
-     * @see Collection::getAggregateOptions() for supported $options
      *
      * @param array $pipeline The pipeline to execute
      * @param array $options  Additional options
      * @return Iterator
      */
     public function aggregate(array $pipeline, array $options = array())
     {
-        $options = array_merge($this->getAggregateOptions(), $options);
-        $options = $this->_massageAggregateOptions($options);
-        $cmd = array(
-            "aggregate" => $this->collname,
-            "pipeline"  => $pipeline,
-        ) + $options;
+        $readPreference = new ReadPreference(ReadPreference::RP_PRIMARY);
+        $server = $this->manager->selectServer($readPreference);
+
+        if (FeatureDetection::isSupported($server, FeatureDetection::API_AGGREGATE_CURSOR)) {
+            $options = array_merge(
+                array(
+                    /**
+                     * Enables writing to temporary files. When set to true, aggregation stages
+                     * can write data to the _tmp subdirectory in the dbPath directory. The
+                     * default is false.
+                     *
+                     * @see http://docs.mongodb.org/manual/reference/command/aggregate/
+                     */
+                    'allowDiskUse' => false,
+                    /**
+                     * The number of documents to return per batch.
+                     *
+                     * @see http://docs.mongodb.org/manual/reference/command/aggregate/
+                     */
+                    'batchSize' => 0,
+                    /**
+                     * The maximum amount of time to allow the query to run.
+                     *
+                     * @see http://docs.mongodb.org/manual/reference/command/aggregate/
+                     */
+                    'maxTimeMS' => 0,
+                    /**
+                     * Indicates if the results should be provided as a cursor.
+                     *
+                     * @see http://docs.mongodb.org/manual/reference/command/aggregate/
+                     */
+                    'useCursor' => true,
+                ),
+                $options
+            );
+        }
 
-        $cursor = $this->_runCommand($this->dbname, $cmd);
+        $options = $this->_massageAggregateOptions($options);
+        $command = new Command(array(
+            'aggregate' => $this->collname,
+            'pipeline' => $pipeline,
+        ) + $options);
+        $cursor = $server->executeCommand($this->dbname, $command);
 
-        if (isset($cmd["cursor"]) && $cmd["cursor"]) {
+        if ( ! empty($options["cursor"])) {
             return $cursor;
         }
 
         $doc = current($cursor->toArray());
 
         if ($doc["ok"]) {
-            return new \ArrayIterator($doc["result"]);
+            return new \ArrayIterator(array_map(
+                function (\stdClass $document) { return (array) $document; },
+                $doc["result"]
+            ));
         }
 
         throw $this->_generateCommandException($doc);
@@ -578,55 +616,6 @@ public function findOneAndUpdate(array $filter, array $update, array $options =
         throw $this->_generateCommandException($doc);
     }
 
-    /**
-     * Retrieves all aggregate options with their default values.
-     *
-     * @return array of Collection::aggregate() options
-     */
-    public function getAggregateOptions()
-    {
-        $opts = array(
-            /**
-             * Enables writing to temporary files. When set to true, aggregation stages
-             * can write data to the _tmp subdirectory in the dbPath directory. The
-             * default is false.
-             *
-             * @see http://docs.mongodb.org/manual/reference/command/aggregate/
-             */
-            "allowDiskUse" => false,
-
-            /**
-             * The number of documents to return per batch.
-             *
-             * @see http://docs.mongodb.org/manual/reference/command/aggregate/
-             */
-            "batchSize" => 0,
-
-            /**
-             * The maximum amount of time to allow the query to run.
-             *
-             * @see http://docs.mongodb.org/manual/reference/command/aggregate/
-             */
-            "maxTimeMS" => 0,
-
-            /**
-             * Indicates if the results should be provided as a cursor.
-             *
-             * The default for this value depends on the version of the server.
-             * - Servers >= 2.6 will use a default of true.
-             * - Servers < 2.6 will use a default of false.
-             *
-             * As with any other property, this value can be changed.
-             *
-             * @see http://docs.mongodb.org/manual/reference/command/aggregate/
-             */
-            "useCursor" => true,
-        );
-
-        /* FIXME: Add a version check for useCursor */
-        return $opts;
-    }
-
     /**
      * Retrieves all Bulk Write options with their default values.
      *
@@ -1151,8 +1140,10 @@ final protected function _generateCommandException($doc)
      */
     protected function _massageAggregateOptions($options)
     {
-        if ($options["useCursor"]) {
-            $options["cursor"] = array("batchSize" => $options["batchSize"]);
+        if ( ! empty($options["useCursor"])) {
+            $options["cursor"] = isset($options["batchSize"])
+                ? array("batchSize" => (integer) $options["batchSize"])
+                : new stdClass;
         }
         unset($options["useCursor"], $options["batchSize"]);
 

src/FeatureDetection.php

@@ -14,6 +14,7 @@ class FeatureDetection
     const API_LISTCOLLECTIONS_CMD = 3;
     const API_LISTINDEXES_CMD = 3;
     const API_CREATEINDEXES_CMD = 2;
+    const API_AGGREGATE_CURSOR = 2;
 
     /**
      * Return whether the server supports a particular feature.

tests/Collection/CrudSpec/AggregateFunctionalTest.php

@@ -33,7 +33,8 @@ public function testAggregateWithMultipleStages()
             array('_id' => 3, 'x' => 33),
         );
 
-        $this->assertSame($expected, $cursor->toArray());
+        // Use iterator_to_array() here since aggregate() may return an ArrayIterator
+        $this->assertSame($expected, iterator_to_array($cursor));
     }
 
     public function testAggregateWithOut()