Merge pull request #29

.travis.yml

@@ -12,9 +12,9 @@ env:
     - MONGO_REPO_TYPE="precise/mongodb-enterprise/"
     - SOURCES_LOC="/etc/apt/sources.list.d/mongodb.list"
   matrix:
-    - DRIVER_VERSION=alpha SERVER_VERSION=2.4
-    - DRIVER_VERSION=alpha SERVER_VERSION=2.6
-    - DRIVER_VERSION=alpha SERVER_VERSION=3.0
+    - DRIVER_VERSION=beta SERVER_VERSION=2.4
+    - DRIVER_VERSION=beta SERVER_VERSION=2.6
+    - DRIVER_VERSION=beta SERVER_VERSION=3.0
 
 before_install:
   - sudo apt-key adv --keyserver ${KEY_SERVER} --recv 7F0CEB10

src/Client.php

@@ -75,8 +75,8 @@ class Client
     public function selectCollection($databaseName, $collectionName, WriteConcern $writeConcern = null, ReadPreference $readPreference = null)
     {
         $namespace = $databaseName . '.' . $collectionName;
-        $writeConcern = $writeConcern ?: $this->manager->getWriteConcern();
-        $readPreference = $readPreference ?: $this->manager->getReadPreference();
+        $writeConcern = $writeConcern ?: \MongoDB\get_manager_write_concern($this->manager);
+        $readPreference = $readPreference ?: \MongoDB\get_manager_read_preference($this->manager);
 
         return new Collection($this->manager, $namespace, $writeConcern, $readPreference);
     }
@@ -94,8 +94,8 @@ class Client
      */
     public function selectDatabase($databaseName, WriteConcern $writeConcern = null, ReadPreference $readPreference = null)
     {
-        $writeConcern = $writeConcern ?: $this->manager->getWriteConcern();
-        $readPreference = $readPreference ?: $this->manager->getReadPreference();
+        $writeConcern = $writeConcern ?: \MongoDB\get_manager_write_concern($this->manager);
+        $readPreference = $readPreference ?: \MongoDB\get_manager_read_preference($this->manager);
 
         return new Database($this->manager, $databaseName, $writeConcern, $readPreference);
     }

src/Collection.php

@@ -66,8 +66,8 @@ class Collection
         $this->collectionName = $parts[1];
 
         $this->manager = $manager;
-        $this->writeConcern = $writeConcern;
-        $this->readPreference = $readPreference;
+        $this->writeConcern = $writeConcern ?: \MongoDB\get_manager_write_concern($this->manager);
+        $this->readPreference = $readPreference ?: \MongoDB\get_manager_read_preference($this->manager);
     }
 
     /**
@@ -95,9 +95,16 @@ class Collection
      */
     public function aggregate(array $pipeline, array $options = array())
     {
-        $readPreference = new ReadPreference(ReadPreference::RP_PRIMARY);
-        $server = $this->manager->selectServer($readPreference);
+        if ( ! isset($options['readPreference'])) {
+            $options['readPreference'] = $this->readPreference;
+        }
+
+        if (\MongoDB\is_last_pipeline_operator_out($pipeline)) {
+            $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
+        }
+
         $operation = new Aggregate($this->databaseName, $this->collectionName, $pipeline, $options);
+        $server = $this->manager->selectServer($options['readPreference']);
 
         return $operation->execute($server);
     }
@@ -112,7 +119,7 @@ class Collection
      */
     public function bulkWrite(array $operations, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -132,8 +139,12 @@ class Collection
      */
     public function count($filter = array(), array $options = array())
     {
+        if ( ! isset($options['readPreference'])) {
+            $options['readPreference'] = $this->readPreference;
+        }
+
         $operation = new Count($this->databaseName, $this->collectionName, $filter, $options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = $this->manager->selectServer($options['readPreference']);
 
         return $operation->execute($server);
     }
@@ -194,7 +205,7 @@ class Collection
      */
     public function deleteMany($filter, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -215,7 +226,7 @@ class Collection
      */
     public function deleteOne($filter, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -236,8 +247,12 @@ class Collection
      */
     public function distinct($fieldName, $filter = array(), array $options = array())
     {
+        if ( ! isset($options['readPreference'])) {
+            $options['readPreference'] = $this->readPreference;
+        }
+
         $operation = new Distinct($this->databaseName, $this->collectionName, $fieldName, $filter, $options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = $this->manager->selectServer($options['readPreference']);
 
         return $operation->execute($server);
     }
@@ -300,8 +315,12 @@ class Collection
      */
     public function find($filter = array(), array $options = array())
     {
+        if ( ! isset($options['readPreference'])) {
+            $options['readPreference'] = $this->readPreference;
+        }
+
         $operation = new Find($this->databaseName, $this->collectionName, $filter, $options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = $this->manager->selectServer($options['readPreference']);
 
         return $operation->execute($server);
     }
@@ -317,8 +336,12 @@ class Collection
      */
     public function findOne($filter = array(), array $options = array())
     {
+        if ( ! isset($options['readPreference'])) {
+            $options['readPreference'] = $this->readPreference;
+        }
+
         $operation = new FindOne($this->databaseName, $this->collectionName, $filter, $options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = $this->manager->selectServer($options['readPreference']);
 
         return $operation->execute($server);
     }
@@ -430,7 +453,7 @@ class Collection
      */
     public function insertMany(array $documents, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -451,7 +474,7 @@ class Collection
      */
     public function insertOne($document, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -487,7 +510,7 @@ class Collection
      */
     public function replaceOne($filter, $replacement, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -509,7 +532,7 @@ class Collection
      */
     public function updateMany($filter, $update, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -531,7 +554,7 @@ class Collection
      */
     public function updateOne($filter, $update, array $options = array())
     {
-        if ( ! isset($options['writeConcern']) && isset($this->writeConcern)) {
+        if ( ! isset($options['writeConcern'])) {
             $options['writeConcern'] = $this->writeConcern;
         }
 

src/Database.php

@@ -44,8 +44,8 @@ class Database
 
         $this->manager = $manager;
         $this->databaseName = (string) $databaseName;
-        $this->writeConcern = $writeConcern;
-        $this->readPreference = $readPreference;
+        $this->writeConcern = $writeConcern ?: \MongoDB\get_manager_write_concern($this->manager);
+        $this->readPreference = $readPreference ?: \MongoDB\get_manager_read_preference($this->manager);
     }
 
     /**

src/Operation/Aggregate.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\InvalidArgumentTypeException;
@@ -42,6 +43,8 @@ class Aggregate implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
      *  * useCursor (boolean): Indicates whether the command will request that
      *    the server provide results using a cursor. The default is true.
      *
@@ -94,6 +97,10 @@ class Aggregate implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
+            throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        }
+
         if ( ! is_bool($options['useCursor'])) {
             throw new InvalidArgumentTypeException('"useCursor" option', $options['useCursor'], 'boolean');
         }
@@ -118,9 +125,10 @@ class Aggregate implements Executable
     public function execute(Server $server)
     {
         $isCursorSupported = \MongoDB\server_supports_feature($server, self::$wireVersionForCursor);
+        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
 
         $command = $this->createCommand($server, $isCursorSupported);
-        $cursor = $server->executeCommand($this->databaseName, $command);
+        $cursor = $server->executeCommand($this->databaseName, $command, $readPreference);
 
         if ($isCursorSupported && $this->options['useCursor']) {
             return $cursor;

src/Operation/BulkWrite.php

@@ -188,7 +188,7 @@ class BulkWrite implements Executable
             throw new InvalidArgumentTypeException('"ordered" option', $options['ordered'], 'boolean');
         }
 
-        if (array_key_exists('writeConcern', $options) && ! $options['writeConcern'] instanceof WriteConcern) {
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
             throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
         }
 

src/Operation/Count.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\InvalidArgumentTypeException;
@@ -36,6 +37,8 @@ class Count implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
      *  * skip (integer): The number of documents to skip before returning the
      *    documents.
      *
@@ -69,6 +72,10 @@ class Count implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
+            throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        }
+
         if (isset($options['skip']) && ! is_integer($options['skip'])) {
             throw new InvalidArgumentTypeException('"skip" option', $options['skip'], 'integer');
         }
@@ -88,7 +95,9 @@ class Count implements Executable
      */
     public function execute(Server $server)
     {
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
+
+        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
         $result = current($cursor->toArray());
 
         if (empty($result->ok)) {

src/Operation/Delete.php

@@ -52,7 +52,7 @@ class Delete implements Executable
             throw new InvalidArgumentException('$limit must be 0 or 1');
         }
 
-        if (array_key_exists('writeConcern', $options) && ! $options['writeConcern'] instanceof WriteConcern) {
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
             throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
         }
 

src/Operation/Distinct.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\InvalidArgumentTypeException;
@@ -32,6 +33,8 @@ class Distinct implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
      * @param string       $databaseName   Database name
      * @param string       $collectionName Collection name
      * @param string       $fieldName      Field for which to return distinct values
@@ -49,6 +52,10 @@ class Distinct implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
+            throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        }
+
         $this->databaseName = (string) $databaseName;
         $this->collectionName = (string) $collectionName;
         $this->fieldName = (string) $fieldName;
@@ -65,7 +72,9 @@ class Distinct implements Executable
      */
     public function execute(Server $server)
     {
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
+
+        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
         $result = current($cursor->toArray());
 
         if (empty($result->ok)) {

src/Operation/Find.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Query;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\InvalidArgumentTypeException;
@@ -64,6 +65,8 @@ class Find implements Executable
      *  * projection (document): Limits the fields to return for the matching
      *    document.
      *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
      *  * skip (integer): The number of documents to skip before returning.
      *
      *  * sort (document): The order in which to return matching documents. If
@@ -130,6 +133,10 @@ class Find implements Executable
             throw new InvalidArgumentTypeException('"projection" option', $options['projection'], 'array or object');
         }
 
+        if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
+            throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        }
+
         if (isset($options['skip']) && ! is_integer($options['skip'])) {
             throw new InvalidArgumentTypeException('"skip" option', $options['skip'], 'integer');
         }
@@ -153,7 +160,9 @@ class Find implements Executable
      */
     public function execute(Server $server)
     {
-        return $server->executeQuery($this->databaseName . '.' . $this->collectionName, $this->createQuery());
+        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
+
+        return $server->executeQuery($this->databaseName . '.' . $this->collectionName, $this->createQuery(), $readPreference);
     }
 
     /**

src/Operation/FindOne.php

@@ -34,6 +34,8 @@ class FindOne implements Executable
      *  * projection (document): Limits the fields to return for the matching
      *    document.
      *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
      *  * skip (integer): The number of documents to skip before returning.
      *
      *  * sort (document): The order in which to return matching documents. If

src/Operation/InsertMany.php

@@ -68,7 +68,7 @@ class InsertMany implements Executable
             throw new InvalidArgumentTypeException('"ordered" option', $options['ordered'], 'boolean');
         }
 
-        if (array_key_exists('writeConcern', $options) && ! $options['writeConcern'] instanceof WriteConcern) {
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
             throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
         }
 

src/Operation/InsertOne.php

@@ -41,7 +41,7 @@ class InsertOne implements Executable
             throw new InvalidArgumentTypeException('$document', $document, 'array or object');
         }
 
-        if (array_key_exists('writeConcern', $options) && ! $options['writeConcern'] instanceof WriteConcern) {
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
             throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
         }
 

src/Operation/Update.php

@@ -75,7 +75,7 @@ class Update implements Executable
             throw new InvalidArgumentTypeException('"upsert" option', $options['upsert'], 'boolean');
         }
 
-        if (array_key_exists('writeConcern', $options) && ! $options['writeConcern'] instanceof WriteConcern) {
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
             throw new InvalidArgumentTypeException('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
         }
 

src/functions.php

@@ -2,8 +2,12 @@
 
 namespace MongoDB;
 
+use MongoDB\Driver\Manager;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
+use MongoDB\Driver\WriteConcern;
 use MongoDB\Exception\InvalidArgumentTypeException;
+use ReflectionClass;
 
 /**
  * Return whether the first key in the document starts with a "$" character.
@@ -30,6 +34,92 @@ function is_first_key_operator($document)
     return (isset($firstKey[0]) && $firstKey[0] == '$');
 }
 
+/**
+ * Return whether the aggregation pipeline ends with an $out operator.
+ *
+ * This is used for determining whether the aggregation pipeline msut be
+ * executed against a primary server.
+ *
+ * @internal
+ * @param array $pipeline List of pipeline operations
+ * @return boolean
+ */
+function is_last_pipeline_operator_out(array $pipeline)
+{
+    $lastOp = end($pipeline);
+
+    if ($lastOp === false) {
+        return false;
+    }
+
+    $lastOp = (array) $lastOp;
+
+    return key($lastOp) === '$out';
+}
+
+/**
+ * Returns a ReadPreference corresponding to the Manager's read preference.
+ *
+ * @internal
+ * @todo this function can be removed once PHPC-417 is implemented
+ * @param Manager $manager
+ * @return ReadPreference
+ */
+function get_manager_read_preference(Manager $manager)
+{
+    $rp = $manager->getReadPreference();
+
+    if ($rp instanceof ReadPreference) {
+        return $rp;
+    }
+
+    $args = array(
+        $rp['mode'],
+    );
+
+    if (isset($rp['tags'])) {
+        $args[] = $rp['tags'];
+    }
+
+    $rc = new ReflectionClass('MongoDB\Driver\ReadPreference');
+
+    return $rc->newInstanceArgs($args);
+}
+
+/**
+ * Returns a WriteConcern corresponding to the Manager's write concern.
+ *
+ * @internal
+ * @todo this function can be removed once PHPC-417 is implemented
+ * @param Manager $manager
+ * @return WriteConcern
+ */
+function get_manager_write_concern(Manager $manager)
+{
+    $wc = $manager->getWriteConcern();
+
+    if ($wc instanceof WriteConcern) {
+        return $wc;
+    }
+
+    $args = array(
+        isset($wc['w']) ? $wc['w'] : -2,
+        $wc['wtimeout'],
+    );
+
+    if (isset($wc['journal'])) {
+        $args[] = $wc['journal'];
+
+        if (isset($wc['fsync'])) {
+            $args[] = $wc['fsync'];
+        }
+    }
+
+    $rc = new ReflectionClass('MongoDB\Driver\WriteConcern');
+
+    return $rc->newInstanceArgs($args);
+}
+
 /**
  * Generate an index name from a key specification.
  *

tests/Operation/AggregateTest.php

@@ -49,6 +49,10 @@ class AggregateTest extends TestCase
             $options[][] = array('maxTimeMS' => $value);
         }
 
+        foreach ($this->getInvalidReadPreferenceValues() as $value) {
+            $options[][] = array('readPreference' => $value);
+        }
+
         foreach ($this->getInvalidBooleanValues() as $value) {
             $options[][] = array('useCursor' => $value);
         }

tests/Operation/CountTest.php

@@ -40,6 +40,10 @@ class CountTest extends TestCase
             $options[][] = array('maxTimeMS' => $value);
         }
 
+        foreach ($this->getInvalidReadPreferenceValues() as $value) {
+            $options[][] = array('readPreference' => $value);
+        }
+
         foreach ($this->getInvalidIntegerValues() as $value) {
             $options[][] = array('skip' => $value);
         }

tests/Operation/DistinctTest.php

@@ -32,6 +32,10 @@ class DistinctTest extends TestCase
             $options[][] = array('maxTimeMS' => $value);
         }
 
+        foreach ($this->getInvalidReadPreferenceValues() as $value) {
+            $options[][] = array('readPreference' => $value);
+        }
+
         return $options;
     }
 }

tests/Operation/FindTest.php

@@ -65,6 +65,10 @@ class FindTest extends TestCase
             $options[][] = array('projection' => $value);
         }
 
+        foreach ($this->getInvalidReadPreferenceValues() as $value) {
+            $options[][] = array('readPreference' => $value);
+        }
+
         foreach ($this->getInvalidIntegerValues() as $value) {
             $options[][] = array('skip' => $value);
         }

tests/Operation/TestCase.php

@@ -40,6 +40,11 @@ abstract class TestCase extends BaseTestCase
         return array(123, 3.14, true, array(), new stdClass);
     }
 
+    protected function getInvalidReadPreferenceValues()
+    {
+        return array(123, 3.14, 'foo', true, array(), new stdClass);
+    }
+
     protected function getInvalidWriteConcernValues()
     {
         return array(123, 3.14, 'foo', true, array(), new stdClass);