Merge pull request #632

Author: jmikola

src/Operation/Watch.php

@@ -47,7 +47,7 @@
  */
 class Watch implements Executable, /* @internal */ CommandSubscriber
 {
-    private static $wireVersionForOperationTime = 7;
+    private static $wireVersionForStartAtOperationTime = 7;
 
     const FULL_DOCUMENT_DEFAULT = 'default';
     const FULL_DOCUMENT_UPDATE_LOOKUP = 'updateLookup';
@@ -267,9 +267,9 @@ private function createResumeCallable(Manager $manager)
     private function executeAggregate(Server $server)
     {
         /* If we've already captured an operation time or the server does not
-         * support returning an operation time (e.g. MongoDB 3.6), execute the
-         * aggregation directly and return its cursor. */
-        if ($this->operationTime !== null || ! \MongoDB\server_supports_feature($server, self::$wireVersionForOperationTime)) {
+         * support resuming from an operation time (e.g. MongoDB 3.6), execute
+         * the aggregation directly and return its cursor. */
+        if ($this->operationTime !== null || ! \MongoDB\server_supports_feature($server, self::$wireVersionForStartAtOperationTime)) {
             return $this->aggregate->execute($server);
         }
 

tests/Operation/WatchFunctionalTest.php

@@ -21,6 +21,8 @@
 
 class WatchFunctionalTest extends FunctionalTestCase
 {
+    private static $wireVersionForStartAtOperationTime = 7;
+
     private $defaultOptions = ['maxAwaitTimeMS' => 500];
 
     public function setUp()
@@ -134,6 +136,8 @@ function(array $event) use (&$commands) {
 
     public function testResumeBeforeReceivingAnyResultsIncludesStartAtOperationTime()
     {
+        $this->skipIfStartAtOperationTimeNotSupported();
+
         $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
 
         $operationTime = null;
@@ -150,7 +154,9 @@ function (array $event) use (&$events) {
 
         $this->assertCount(1, $events);
         $this->assertSame('aggregate', $events[0]['started']->getCommandName());
-        $operationTime = $events[0]['succeeded']->getReply()->operationTime;
+        $reply = $events[0]['succeeded']->getReply();
+        $this->assertObjectHasAttribute('operationTime', $reply);
+        $operationTime = $reply->operationTime;
         $this->assertInstanceOf(TimestampInterface::class, $operationTime);
 
         $this->assertNull($changeStream->current());
@@ -398,20 +404,29 @@ public function testResumeMultipleTimesInSuccession()
 
         $this->insertDocument(['_id' => 1]);
 
+        /* Insert a document and advance the change stream to ensure we capture
+         * a resume token. This is necessary when startAtOperationTime is not
+         * supported (i.e. 3.6 server version). */
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $this->assertSame(0, $changeStream->key());
+
+        $this->insertDocument(['_id' => 2]);
+
         /* Killing the cursor and advancing when there is a result will test
          * that next()'s resume attempt picks up the latest change. */
         $this->killChangeStreamCursor($changeStream);
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
-        $this->assertSame(0, $changeStream->key());
+        $this->assertSame(1, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
             'operationType' => 'insert',
-            'fullDocument' => ['_id' => 1],
+            'fullDocument' => ['_id' => 2],
             'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 1],
+            'documentKey' => ['_id' => 2],
         ];
 
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
@@ -424,48 +439,48 @@ public function testResumeMultipleTimesInSuccession()
 
         $changeStream->rewind();
         $this->assertTrue($changeStream->valid());
-        $this->assertSame(0, $changeStream->key());
+        $this->assertSame(1, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
             'operationType' => 'insert',
-            'fullDocument' => ['_id' => 1],
+            'fullDocument' => ['_id' => 2],
             'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 1],
+            'documentKey' => ['_id' => 2],
         ];
 
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
 
-        $this->insertDocument(['_id' => 2]);
+        $this->insertDocument(['_id' => 3]);
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
-        $this->assertSame(1, $changeStream->key());
+        $this->assertSame(2, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
             'operationType' => 'insert',
-            'fullDocument' => ['_id' => 2],
+            'fullDocument' => ['_id' => 3],
             'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 2],
+            'documentKey' => ['_id' => 3],
         ];
 
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
 
         $this->killChangeStreamCursor($changeStream);
 
-        $this->insertDocument(['_id' => 3]);
+        $this->insertDocument(['_id' => 4]);
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
-        $this->assertSame(2, $changeStream->key());
+        $this->assertSame(3, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
             'operationType' => 'insert',
-            'fullDocument' => ['_id' => 3],
+            'fullDocument' => ['_id' => 4],
             'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 3],
+            'documentKey' => ['_id' => 4],
         ];
 
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
@@ -476,18 +491,18 @@ public function testResumeMultipleTimesInSuccession()
          * we'll see {_id: 3} returned again. */
         $this->killChangeStreamCursor($changeStream);
 
-        $this->insertDocument(['_id' => 4]);
+        $this->insertDocument(['_id' => 5]);
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
-        $this->assertSame(3, $changeStream->key());
+        $this->assertSame(4, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
             'operationType' => 'insert',
-            'fullDocument' => ['_id' => 4],
+            'fullDocument' => ['_id' => 5],
             'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 4],
+            'documentKey' => ['_id' => 5],
         ];
 
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
@@ -944,4 +959,11 @@ private function killChangeStreamCursor(ChangeStream $changeStream)
         $operation = new DatabaseCommand($this->getDatabaseName(), $command);
         $operation->execute($this->getPrimaryServer());
     }
+
+    private function skipIfStartAtOperationTimeNotSupported()
+    {
+        if (!\MongoDB\server_supports_feature($this->getPrimaryServer(), self::$wireVersionForStartAtOperationTime)) {
+             $this->markTestSkipped('Operation time is not supported');
+        }
+    }
 }