PHPLIB-111: Ensure read ops use appropriate read preference

.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 @@ public function listDatabases(array $options = array())
     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 @@ public function selectCollection($databaseName, $collectionName, WriteConcern $w
      */
     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 @@ public function __construct(Manager $manager, $namespace, WriteConcern $writeCon
         $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 @@ public function __toString()
      */
     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 @@ public function aggregate(array $pipeline, array $options = array())
      */
     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 @@ public function bulkWrite(array $operations, array $options = array())
      */
     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 @@ public function createIndexes(array $indexes)
      */
     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 @@ public function deleteMany($filter, array $options = array())
      */
     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 @@ public function deleteOne($filter, array $options = array())
      */
     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 @@ public function dropIndexes()
      */
     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 @@ public function find($filter = array(), array $options = array())
      */
     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 @@ public function getNamespace()
      */
     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 @@ public function insertMany(array $documents, array $options = array())
      */
     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 @@ public function listIndexes(array $options = array())
      */
     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 @@ public function replaceOne($filter, $replacement, array $options = array())
      */
     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 @@ public function updateMany($filter, $update, array $options = array())
      */
     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 @@ public function __construct(Manager $manager, $databaseName, WriteConcern $write
 
         $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 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
             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 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
     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 @@ public function __construct($databaseName, $collectionName, array $operations, a
             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 @@ public function __construct($databaseName, $collectionName, $filter = array(), a
             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 @@ public function __construct($databaseName, $collectionName, $filter = array(), a
      */
     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 @@ public function __construct($databaseName, $collectionName, $filter, $limit, arr
             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 @@ public function __construct($databaseName, $collectionName, $fieldName, $filter
             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 @@ public function __construct($databaseName, $collectionName, $fieldName, $filter
      */
     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 @@ public function __construct($databaseName, $collectionName, $filter, array $opti
             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 @@ public function __construct($databaseName, $collectionName, $filter, array $opti
      */
     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 @@ public function __construct($databaseName, $collectionName, array $documents, ar
             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');
         }