Merge pull request #476

docs/includes/apiargs-MongoDBCollection-method-watch-option.yaml

-source:
-  file: apiargs-MongoDBCollection-method-aggregate-option.yaml
-  ref: batchSize
+---
+arg_name: option
+name: batchSize
+type: integer
+description: |
+  Specifies the maximum number of change events to return in each batch of the
+  response from the MongoDB cluster.
+interface: phpmethod
+operation: ~
+optional: true
 ---
 source:
   file: apiargs-MongoDBCollection-common-option.yaml
@@ -18,6 +25,10 @@ description: |
 
   - ``MongoDB\Operation\ChangeStream::FULL_DOCUMENT_DEFAULT`` (*default*)
   - ``MongoDB\Operation\ChangeStream::FULL_DOCUMENT_UPDATE_LOOKUP``
+
+  .. note::
+
+     This is an option of the `$changeStream` pipeline stage.
 interface: phpmethod
 operation: ~
 optional: true
@@ -45,7 +56,9 @@ type: array|object
 description: |
   Specifies the logical starting point for the new change stream.
 
-  Note this is an option of the '$changeStream' pipeline stage.
+  .. note::
+
+     This is an option of the `$changeStream` pipeline stage.
 interface: phpmethod
 operation: ~
 optional: true

src/Collection.php

@@ -19,7 +19,7 @@ namespace MongoDB;
 
 use MongoDB\BSON\JavascriptInterface;
 use MongoDB\BSON\Serializable;
-use MongoDB\ChangeStream as ChangeStreamResult;
+use MongoDB\ChangeStream;
 use MongoDB\Driver\Cursor;
 use MongoDB\Driver\Manager;
 use MongoDB\Driver\ReadConcern;
@@ -32,7 +32,6 @@ use MongoDB\Exception\UnsupportedException;
 use MongoDB\Model\IndexInfoIterator;
 use MongoDB\Operation\Aggregate;
 use MongoDB\Operation\BulkWrite;
-use MongoDB\Operation\ChangeStream;
 use MongoDB\Operation\CreateIndexes;
 use MongoDB\Operation\Count;
 use MongoDB\Operation\DeleteMany;
@@ -52,6 +51,7 @@ use MongoDB\Operation\MapReduce;
 use MongoDB\Operation\ReplaceOne;
 use MongoDB\Operation\UpdateMany;
 use MongoDB\Operation\UpdateOne;
+use MongoDB\Operation\Watch;
 use Traversable;
 
 class Collection
@@ -939,14 +939,14 @@ class Collection
         return $operation->execute($server);
     }
 
-    /*
-     * ChangeStream outline
+    /**
+     * Create a change stream for watching changes to the collection.
      *
-     * @see ChangeStream::__construct() for supported options
+     * @see Watch::__construct() for supported options
      * @param array $pipeline List of pipeline operations
      * @param array $options  Command options
+     * @return ChangeStream
      * @throws InvalidArgumentException for parameter/option parsing errors
-     * @return ChangeStreamResult
      */
     public function watch(array $pipeline = [], array $options = [])
     {
@@ -956,11 +956,18 @@ class Collection
 
         $server = $this->manager->selectServer($options['readPreference']);
 
-        if ( ! isset($options['readConcern'])) {
+        /* Although change streams require a newer version of the server than
+         * read concerns, perform the usual wire version check before inheriting
+         * the collection's read concern. In the event that the server is too
+         * old, this makes it more likely that users will encounter an error
+         * related to change streams being unsupported instead of an
+         * UnsupportedException regarding use of the "readConcern" option from
+         * the Aggregate operation class. */
+        if ( ! isset($options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
             $options['readConcern'] = $this->readConcern;
         }
 
-        $operation = new ChangeStream($this->databaseName, $this->collectionName, $pipeline, $options, $this->manager);
+        $operation = new Watch($this->manager, $this->databaseName, $this->collectionName, $pipeline, $options);
 
         return $operation->execute($server);
     }

src/Operation/ChangeStream.php → src/Operation/Watch.php

@@ -17,7 +17,7 @@
 
 namespace MongoDB\Operation;
 
-use MongoDB\ChangeStream as ChangeStreamResult;
+use MongoDB\ChangeStream;
 use MongoDB\Driver\Command;
 use MongoDB\Driver\Manager;
 use MongoDB\Driver\ReadConcern;
@@ -32,10 +32,10 @@ use MongoDB\Exception\UnsupportedException;
  * Operation for creating a change stream with the aggregate command.
  *
  * @api
- * @see \MongoDB\Collection::changeStream()
- * @see http://docs.mongodb.org/manual/reference/command/changeStream/
+ * @see \MongoDB\Collection::watch()
+ * @see https://docs.mongodb.com/manual/changeStreams/
  */
-class ChangeStream implements Executable
+class Watch implements Executable
 {
     const FULL_DOCUMENT_DEFAULT = 'default';
     const FULL_DOCUMENT_UPDATE_LOOKUP = 'updateLookup';
@@ -47,18 +47,19 @@ class ChangeStream implements Executable
     private $manager;
 
     /**
-     * Constructs a changeStream command.
+     * Constructs an aggregate command for creating a change stream.
      *
      * Supported options:
      *
-     *  * fullDocument (string): Allowed values: ‘default’, ‘updateLookup’.
-     *    Defaults to ‘default’.  When set to ‘updateLookup’, the change
-     *    notification for partial updates will include both a delta describing
-     *    the changes to the document, as well as a copy of the entire document
-     *    that was changed from some time after the change occurred. For forward
-     *    compatibility, a driver MUST NOT raise an error when a user provides
-     *    an unknown value. The driver relies on the server to validate this
-     *    option.
+     *  * fullDocument (string): Determines whether the "fullDocument" field
+     *    will be populated for update operations. By default, change streams
+     *    only return the delta of fields during the update operation (via the
+     *    "updateDescription" field). To additionally return the most current
+     *    majority-committed version of the updated document, specify
+     *    "updateLookup" for this option. Defaults to "default".
+     *
+     *    Insert and replace operations always include the "fullDocument" field
+     *    and delete operations omit the field as the document no longer exists.
      *
      *  * resumeAfter (document): Specifies the logical starting point for the
      *    new change stream.
@@ -69,7 +70,9 @@ class ChangeStream implements Executable
      *    This is not supported for server versions < 3.2 and will result in an
      *    exception at execution time if used.
      *
-     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference. This
+     *    will be used to select a new server when resuming. Defaults to a
+     *    "primary" read preference.
      *
      *  * maxAwaitTimeMS (integer): The maximum amount of time for the server to
      *    wait on new documents to satisfy a change stream query.
@@ -91,8 +94,13 @@ class ChangeStream implements Executable
      * @param Manager        $manager        Manager instance from the driver
      * @throws InvalidArgumentException for parameter/option parsing errors
      */
-    public function __construct($databaseName, $collectionName, array $pipeline, array $options = [], Manager $manager)
+    public function __construct(Manager $manager, $databaseName, $collectionName, array $pipeline, array $options = [])
     {
+        $options += [
+            'fullDocument' => self::FULL_DOCUMENT_DEFAULT,
+            'readPreference' => new ReadPreference(ReadPreference::RP_PRIMARY),
+        ];
+
         if (isset($options['batchSize']) && ! is_integer($options['batchSize'])) {
             throw InvalidArgumentException::invalidType('"batchSize" option', $options['batchSize'], 'integer');
         }
@@ -119,11 +127,11 @@ class ChangeStream implements Executable
             }
         }
 
+        $this->manager = $manager;
         $this->databaseName = (string) $databaseName;
         $this->collectionName = (string) $collectionName;
         $this->pipeline = $pipeline;
         $this->options = $options;
-        $this->manager = $manager;
     }
 
     /**
@@ -131,7 +139,7 @@ class ChangeStream implements Executable
      *
      * @see Executable::execute()
      * @param Server $server
-     * @return ChangeStreamResult
+     * @return ChangeStream
      * @throws UnexpectedValueException if the command response was malformed
      * @throws UnsupportedException if collation, read concern, or write concern is used and unsupported
      * @throws DriverRuntimeException for other driver errors (e.g. connection errors)
@@ -142,7 +150,7 @@ class ChangeStream implements Executable
 
         $cursor = $command->execute($server);
 
-        return new ChangeStreamResult($cursor, $this->createResumeCallable());
+        return new ChangeStream($cursor, $this->createResumeCallable());
     }
 
     private function createAggregateOptions()

tests/CommandObserver.php

@@ -6,6 +6,7 @@ use MongoDB\Driver\Monitoring\CommandFailedEvent;
 use MongoDB\Driver\Monitoring\CommandStartedEvent;
 use MongoDB\Driver\Monitoring\CommandSucceededEvent;
 use MongoDB\Driver\Monitoring\CommandSubscriber;
+use Exception;
 
 /**
  * Observes command documents using the driver's monitoring API.
@@ -20,13 +21,19 @@ class CommandObserver implements CommandSubscriber
 
         \MongoDB\Driver\Monitoring\addSubscriber($this);
 
+        try {
             call_user_func($execution);
+        } catch (Exception $executionException) {}
 
         \MongoDB\Driver\Monitoring\removeSubscriber($this);
 
         foreach ($this->commands as $command) {
             call_user_func($commandCallback, $command);
         }
+
+        if (isset($executionException)) {
+            throw $executionException;
+        }
     }
 
     public function commandStarted(CommandStartedEvent $event)

tests/DocumentationExamplesTest.php

@@ -4,6 +4,7 @@ namespace MongoDB\Tests;
 
 use MongoDB\Database;
 use MongoDB\Driver\Cursor;
+use MongoDB\Driver\Server;
 use MongoDB\Operation\DropCollection;
 use MongoDB\Operation\DropDatabase;
 
@@ -923,66 +924,102 @@ class DocumentationExamplesTest extends FunctionalTestCase
 
     public function testChangeStreamExample_1_4()
     {
+        if ($this->getPrimaryServer()->getType() === Server::TYPE_STANDALONE) {
+            $this->markTestSkipped('$changeStream is not supported on standalone servers');
+        }
+
         if (version_compare($this->getFeatureCompatibilityVersion(), '3.6', '<')) {
             $this->markTestSkipped('$changeStream is only supported on FCV 3.6 or higher');
         }
 
         $db = new Database($this->manager, $this->getDatabaseName());
+        $db->dropCollection('inventory');
 
         // Start Changestream Example 1
-        $cursor = $db->inventory->watch();
-        $cursor->next();
-        $current = $cursor->current();
+        $changeStream = $db->inventory->watch();
+        $changeStream->rewind();
+
+        $firstChange = $changeStream->current();
+
+        $changeStream->next();
+
+        $secondChange = $changeStream->current();
         // End Changestream Example 1
 
-        $this->assertNull($current);
+        $this->assertNull($firstChange);
+        $this->assertNull($secondChange);
 
         // Start Changestream Example 2
-        $cursor = $db->inventory->watch([], ['fullDocument' => \MongoDB\Operation\ChangeStream::FULL_DOCUMENT_UPDATE_LOOKUP]);
-        $cursor->next();
-        $current = $cursor->current();
+        $changeStream = $db->inventory->watch([], ['fullDocument' => \MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP]);
+        $changeStream->rewind();
+
+        $firstChange = $changeStream->current();
+
+        $changeStream->next();
+
+        $nextChange = $changeStream->current();
         // End Changestream Example 2
 
-        $this->assertNull($current);
+        $this->assertNull($firstChange);
+        $this->assertNull($nextChange);
+
+        $insertManyResult = $db->inventory->insertMany([
+            ['_id' => 1, 'x' => 'foo'],
+            ['_id' => 2, 'x' => 'bar'],
+        ]);
+        $this->assertEquals(2, $insertManyResult->getInsertedCount());
 
-        $insertedResult = $db->inventory->insertOne(['x' => 1]);
-        $insertedId = $insertedResult->getInsertedId();
-        $cursor->next();
-        $current = $cursor->current();
-        $expectedChange = (object) [
-            '_id' => $current->_id,
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $lastChange = $changeStream->current();
+
+        $expectedChange = [
+            '_id' => $lastChange->_id,
             'operationType' => 'insert',
-            'fullDocument' => (object) ['_id' => $insertedId, 'x' => 1],
-            'ns' => (object) ['db' => 'phplib_test', 'coll' => 'inventory'],
-            'documentKey' => (object) ['_id' => $insertedId]
+            'fullDocument' => ['_id' => 1, 'x' => 'foo'],
+            'ns' => ['db' => $this->getDatabaseName(), 'coll' => 'inventory'],
+            'documentKey' => ['_id' => 1],
         ];
-        $this->assertEquals($current, $expectedChange);
+
+        $this->assertSameDocument($expectedChange, $lastChange);
 
         // Start Changestream Example 3
-        $resumeToken = ($current !== null) ? $current->_id : null;
-        if ($resumeToken !== null) {
-            $cursor = $db->inventory->watch([], ['resumeAfter' => $resumeToken]);
-            $cursor->next();
+        $resumeToken = ($lastChange !== null) ? $lastChange->_id : null;
+
+        if ($resumeToken === null) {
+            throw new \Exception('resumeToken was not found');
         }
+
+        $changeStream = $db->inventory->watch([], ['resumeAfter' => $resumeToken]);
+        $changeStream->rewind();
+
+        $nextChange = $changeStream->current();
         // End Changestream Example 3
 
-        $insertedResult = $db->inventory->insertOne(['x' => 2]);
-        $insertedId = $insertedResult->getInsertedId();
-        $cursor->next();
-        $expectedChange = (object) [
-            '_id' => $cursor->current()->_id,
+        $expectedChange = [
+            '_id' => $nextChange->_id,
             'operationType' => 'insert',
-            'fullDocument' => (object) ['_id' => $insertedId, 'x' => 2],
-            'ns' => (object) ['db' => 'phplib_test', 'coll' => 'inventory'],
-            'documentKey' => (object) ['_id' => $insertedId]
+            'fullDocument' => ['_id' => 2, 'x' => 'bar'],
+            'ns' => ['db' => $this->getDatabaseName(), 'coll' => 'inventory'],
+            'documentKey' => ['_id' => 2],
         ];
-        $this->assertEquals($cursor->current(), $expectedChange);
+
+        $this->assertSameDocument($expectedChange, $nextChange);
 
         // Start Changestream Example 4
         $pipeline = [['$match' => ['$or' => [['fullDocument.username' => 'alice'], ['operationType' => 'delete']]]]];
-        $cursor = $db->inventory->watch($pipeline, []);
-        $cursor->next();
+        $changeStream = $db->inventory->watch($pipeline);
+        $changeStream->rewind();
+
+        $firstChange = $changeStream->current();
+
+        $changeStream->next();
+
+        $nextChange = $changeStream->current();
         // End Changestream Example 4
+
+        $this->assertNull($firstChange);
+        $this->assertNull($nextChange);
     }
 
     /**