PHPLIB-130: Support readConcern option on read operations

src/Collection.php

@@ -143,11 +143,20 @@ class Collection
      */
     public function aggregate(array $pipeline, array $options = [])
     {
+        $hasOutStage = \MongoDB\is_last_pipeline_operator_out($pipeline);
+
+        /* A "majority" read concern is not compatible with the $out stage, so
+         * avoid providing the Collection's read concern if it would conflict.
+         */
+        if ( ! isset($options['readConcern']) && ! ($hasOutStage && $this->readConcern->getLevel() === ReadConcern::MAJORITY)) {
+            $options['readConcern'] = $this->readConcern;
+        }
+
         if ( ! isset($options['readPreference'])) {
             $options['readPreference'] = $this->readPreference;
         }
 
-        if (\MongoDB\is_last_pipeline_operator_out($pipeline)) {
+        if ($hasOutStage) {
             $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
         }
 
@@ -187,6 +196,10 @@ class Collection
      */
     public function count($filter = [], array $options = [])
     {
+        if ( ! isset($options['readConcern'])) {
+            $options['readConcern'] = $this->readConcern;
+        }
+
         if ( ! isset($options['readPreference'])) {
             $options['readPreference'] = $this->readPreference;
         }
@@ -295,6 +308,10 @@ class Collection
      */
     public function distinct($fieldName, $filter = [], array $options = [])
     {
+        if ( ! isset($options['readConcern'])) {
+            $options['readConcern'] = $this->readConcern;
+        }
+
         if ( ! isset($options['readPreference'])) {
             $options['readPreference'] = $this->readPreference;
         }
@@ -363,6 +380,10 @@ class Collection
      */
     public function find($filter = [], array $options = [])
     {
+        if ( ! isset($options['readConcern'])) {
+            $options['readConcern'] = $this->readConcern;
+        }
+
         if ( ! isset($options['readPreference'])) {
             $options['readPreference'] = $this->readPreference;
         }
@@ -384,6 +405,10 @@ class Collection
      */
     public function findOne($filter = [], array $options = [])
     {
+        if ( ! isset($options['readConcern'])) {
+            $options['readConcern'] = $this->readConcern;
+        }
+
         if ( ! isset($options['readPreference'])) {
             $options['readPreference'] = $this->readPreference;
         }

src/Operation/Aggregate.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
@@ -23,6 +24,7 @@ class Aggregate implements Executable
 {
     private static $wireVersionForCursor = 2;
     private static $wireVersionForDocumentLevelValidation = 4;
+    private static $wireVersionForReadConcern = 4;
 
     private $databaseName;
     private $collectionName;
@@ -50,6 +52,12 @@ class Aggregate implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern. Note that a
+     *    "majority" read concern is not compatible with the $out stage.
+     *
+     *    For servers < 3.2, this option is ignored as read concern is not
+     *    available.
+     *
      *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
      *
      *  * useCursor (boolean): Indicates whether the command will request that
@@ -108,6 +116,10 @@ class Aggregate implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readConcern']) && ! $options['readConcern'] instanceof ReadConcern) {
+            throw new InvalidArgumentTypeException('"readConcern" option', $options['readConcern'], 'MongoDB\Driver\ReadConcern');
+        }
+
         if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
             throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
         }
@@ -183,6 +195,10 @@ class Aggregate implements Executable
             $cmd['maxTimeMS'] = $this->options['maxTimeMS'];
         }
 
+        if (isset($this->options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
+            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
+        }
+
         if ($this->options['useCursor']) {
             $cmd['cursor'] = isset($this->options["batchSize"])
                 ? ['batchSize' => $this->options["batchSize"]]

src/Operation/Count.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
@@ -18,6 +19,8 @@ use MongoDB\Exception\UnexpectedValueException;
  */
 class Count implements Executable
 {
+    private static $wireVersionForReadConcern = 4;
+
     private $databaseName;
     private $collectionName;
     private $filter;
@@ -36,6 +39,11 @@ class Count implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern.
+     *
+     *    For servers < 3.2, this option is ignored as read concern is not
+     *    available.
+     *
      *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
      *
      *  * skip (integer): The number of documents to skip before returning the
@@ -71,6 +79,10 @@ class Count implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readConcern']) && ! $options['readConcern'] instanceof ReadConcern) {
+            throw new InvalidArgumentTypeException('"readConcern" option', $options['readConcern'], 'MongoDB\Driver\ReadConcern');
+        }
+
         if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
             throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
         }
@@ -96,7 +108,7 @@ class Count implements Executable
     {
         $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
+        $cursor = $server->executeCommand($this->databaseName, $this->createCommand($server), $readPreference);
         $result = current($cursor->toArray());
 
         // Older server versions may return a float
@@ -110,9 +122,10 @@ class Count implements Executable
     /**
      * Create the count command.
      *
+     * @param Server $server
      * @return Command
      */
-    private function createCommand()
+    private function createCommand(Server $server)
     {
         $cmd = ['count' => $this->collectionName];
 
@@ -126,6 +139,10 @@ class Count implements Executable
             }
         }
 
+        if (isset($this->options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
+            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
+        }
+
         return new Command($cmd);
     }
 }

src/Operation/Distinct.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
@@ -18,6 +19,8 @@ use MongoDB\Exception\UnexpectedValueException;
  */
 class Distinct implements Executable
 {
+    private static $wireVersionForReadConcern = 4;
+
     private $databaseName;
     private $collectionName;
     private $fieldName;
@@ -32,6 +35,11 @@ class Distinct implements Executable
      *  * maxTimeMS (integer): The maximum amount of time to allow the query to
      *    run.
      *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern.
+     *
+     *    For servers < 3.2, this option is ignored as read concern is not
+     *    available.
+     *
      *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
      *
      * @param string       $databaseName   Database name
@@ -51,6 +59,10 @@ class Distinct implements Executable
             throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
 
+        if (isset($options['readConcern']) && ! $options['readConcern'] instanceof ReadConcern) {
+            throw new InvalidArgumentTypeException('"readConcern" option', $options['readConcern'], 'MongoDB\Driver\ReadConcern');
+        }
+
         if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
             throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
         }
@@ -73,7 +85,7 @@ class Distinct implements Executable
     {
         $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
+        $cursor = $server->executeCommand($this->databaseName, $this->createCommand($server), $readPreference);
         $result = current($cursor->toArray());
 
         if ( ! isset($result->values) || ! is_array($result->values)) {
@@ -86,9 +98,10 @@ class Distinct implements Executable
     /**
      * Create the distinct command.
      *
+     * @param Server $server
      * @return Command
      */
-    private function createCommand()
+    private function createCommand(Server $server)
     {
         $cmd = [
             'distinct' => $this->collectionName,
@@ -103,6 +116,10 @@ class Distinct implements Executable
             $cmd['maxTimeMS'] = $this->options['maxTimeMS'];
         }
 
+        if (isset($this->options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
+            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
+        }
+
         return new Command($cmd);
     }
 }

src/Operation/Find.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Operation;
 
 use MongoDB\Driver\Query;
+use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
@@ -65,6 +66,11 @@ class Find implements Executable
      *  * projection (document): Limits the fields to return for the matching
      *    document.
      *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern.
+     *
+     *    For servers < 3.2, this option is ignored as read concern is not
+     *    available.
+     *
      *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
      *
      *  * skip (integer): The number of documents to skip before returning.
@@ -133,6 +139,10 @@ class Find implements Executable
             throw new InvalidArgumentTypeException('"projection" option', $options['projection'], 'array or object');
         }
 
+        if (isset($options['readConcern']) && ! $options['readConcern'] instanceof ReadConcern) {
+            throw new InvalidArgumentTypeException('"readConcern" option', $options['readConcern'], 'MongoDB\Driver\ReadConcern');
+        }
+
         if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
             throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
         }
@@ -188,7 +198,7 @@ class Find implements Executable
             }
         }
 
-        foreach (['batchSize', 'limit', 'skip', 'sort', 'noCursorTimeout', 'oplogReplay', 'projection'] as $option) {
+        foreach (['batchSize', 'limit', 'skip', 'sort', 'noCursorTimeout', 'oplogReplay', 'projection', 'readConcern'] as $option) {
             if (isset($this->options[$option])) {
                 $options[$option] = $this->options[$option];
             }

src/Operation/FindOne.php

@@ -36,6 +36,11 @@ class FindOne implements Executable
      *  * projection (document): Limits the fields to return for the matching
      *    document.
      *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern.
+     *
+     *    For servers < 3.2, this option is ignored as read concern is not
+     *    available.
+     *
      *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
      *
      *  * skip (integer): The number of documents to skip before returning.

src/functions.php

@@ -2,8 +2,10 @@
 
 namespace MongoDB;
 
+use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentTypeException;
+use stdClass;
 
 /**
  * Return whether the first key in the document starts with a "$" character.
@@ -81,6 +83,25 @@ function generate_index_name($document)
     return $name;
 }
 
+/**
+ * Converts a ReadConcern instance to a stdClass for use in a BSON document.
+ *
+ * @internal
+ * @see https://jira.mongodb.org/browse/PHPC-498
+ * @param ReadConcern $readConcern Read concern
+ * @return stdClass
+ */
+function read_concern_as_document(ReadConcern $readConcern)
+{
+    $document = [];
+
+    if ($readConcern->getLevel() !== null) {
+        $document['level'] = $readConcern->getLevel();
+    }
+
+    return (object) $document;
+}
+
 /**
  * Return whether the server supports a particular feature.
  *

tests/FunctionsTest.php

+<?php
+
+namespace MongoDB\Tests;
+
+use MongoDB\Driver\ReadConcern;
+
+/**
+ * Unit tests for utility functions.
+ */
+class FunctionsTest extends \PHPUnit_Framework_TestCase
+{
+    /**
+     * @dataProvider provideReadConcernsAndDocuments
+     */
+    public function testReadConcernAsDocument(ReadConcern $readConcern, $expectedDocument)
+    {
+        $this->assertEquals($expectedDocument, \MongoDB\read_concern_as_document($readConcern));
+    }
+
+    public function provideReadConcernsAndDocuments()
+    {
+        return [
+            [ new ReadConcern, (object) [] ],
+            [ new ReadConcern(ReadConcern::LOCAL), (object) ['level' => ReadConcern::LOCAL] ],
+            [ new ReadConcern(ReadConcern::MAJORITY), (object) ['level' => ReadConcern::MAJORITY] ],
+        ];
+    }
+}

tests/Operation/AggregateTest.php

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

tests/Operation/CountTest.php

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

tests/Operation/DistinctTest.php

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

tests/Operation/FindTest.php

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

tests/TestCase.php

@@ -2,6 +2,9 @@
 
 namespace MongoDB\Tests;
 
+use MongoDB\Driver\ReadConcern;
+use MongoDB\Driver\ReadPreference;
+use MongoDB\Driver\WriteConcern;
 use ReflectionClass;
 use stdClass;
 
@@ -74,6 +77,16 @@ abstract class TestCase extends \PHPUnit_Framework_TestCase
         return [3.14, 'foo', true, [], new stdClass];
     }
 
+    /**
+     * Return a list of invalid ReadPreference values.
+     *
+     * @return array
+     */
+    protected function getInvalidReadConcernValues()
+    {
+        return [123, 3.14, 'foo', true, [], new stdClass, new ReadPreference(ReadPreference::RP_PRIMARY), new WriteConcern(1)];
+    }
+
     /**
      * Return a list of invalid ReadPreference values.
      *
@@ -81,7 +94,7 @@ abstract class TestCase extends \PHPUnit_Framework_TestCase
      */
     protected function getInvalidReadPreferenceValues()
     {
-        return [123, 3.14, 'foo', true, [], new stdClass];
+        return [123, 3.14, 'foo', true, [], new stdClass, new ReadConcern, new WriteConcern(1)];
     }
 
     /**
@@ -101,7 +114,7 @@ abstract class TestCase extends \PHPUnit_Framework_TestCase
      */
     protected function getInvalidWriteConcernValues()
     {
-        return [123, 3.14, 'foo', true, [], new stdClass];
+        return [123, 3.14, 'foo', true, [], new stdClass, new ReadConcern, new ReadPreference(ReadPreference::RP_PRIMARY)];
     }
 
     /**