Merge pull request #476

Author: jmikola

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

@@ -1,6 +1,13 @@
-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/ChangeStream.php

@@ -100,7 +100,7 @@ public function next()
                 $resumable = true;
             }
             if ($e->getCode() === self::CURSOR_NOT_FOUND) {
-               $resumable = true;
+                $resumable = true;
             }
             if ($e instanceof ConnectionTimeoutException) {
                 $resumable = true;
@@ -147,7 +147,7 @@ private function extractResumeToken($document)
         if (isset($document->_id)) {
             $this->resumeToken = is_array($document) ? $document['_id'] : $document->_id;
         } else {
-             throw new ResumeTokenException("Cannot provide resume functionality when the resume token is missing");
+            throw new ResumeTokenException("Cannot provide resume functionality when the resume token is missing");
         }
     }
 

src/Collection.php

@@ -19,7 +19,7 @@
 
 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\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\ReplaceOne;
 use MongoDB\Operation\UpdateMany;
 use MongoDB\Operation\UpdateOne;
+use MongoDB\Operation\Watch;
 use Traversable;
 
 class Collection
@@ -939,14 +939,14 @@ public function updateOne($filter, $update, array $options = [])
         return $operation->execute($server);
     }
 
-    /*
-     * ChangeStream outline
+    /**
+     * Create a change stream for watching changes to the collection.
      *
-     * @see ChangeStream::__construct() for supported options
-     * @param array          $pipeline       List of pipeline operations
-     * @param array          $options        Command 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 @@ public function watch(array $pipeline = [], array $options = [])
 
         $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 renamed to 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 @@
  * 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,19 +127,19 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
             }
         }
 
+        $this->manager = $manager;
         $this->databaseName = (string) $databaseName;
         $this->collectionName = (string) $collectionName;
         $this->pipeline = $pipeline;
         $this->options = $options;
-        $this->manager = $manager;
     }
 
     /**
      * Execute the operation.
      *
      * @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 @@ public function execute(Server $server)
 
         $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\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 @@ public function observe(callable $execution, callable $commandCallback)
 
         \MongoDB\Driver\Monitoring\addSubscriber($this);
 
-        call_user_func($execution);
+        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)