PHPLIB-411: Fix resuming change stream if startAfter option was passed to original watch operation

Author: alcaeus

src/ChangeStream.php

@@ -220,7 +220,7 @@ private function onIteration($incrementKey)
      */
     private function resume()
     {
-        $this->iterator = call_user_func($this->resumeCallable, $this->getResumeToken());
+        $this->iterator = call_user_func($this->resumeCallable, $this->getResumeToken(), $this->hasAdvanced);
         $this->iterator->rewind();
 
         $this->onIteration($this->hasAdvanced);

src/Operation/Watch.php

@@ -250,7 +250,7 @@ public function execute(Server $server)
     {
         return new ChangeStream(
             $this->createChangeStreamIterator($server),
-            function($resumeToken) { return $this->resume($resumeToken); }
+            function($resumeToken, $hasAdvanced) { return $this->resume($resumeToken, $hasAdvanced); }
         );
     }
 
@@ -333,10 +333,11 @@ private function getInitialResumeToken()
      *
      * @see https://github.com/mongodb/specifications/blob/master/source/change-streams/change-streams.rst#resume-process
      * @param array|object|null $resumeToken
+     * @param bool              $hasAdvanced
      * @return ChangeStreamIterator
      * @throws InvalidArgumentException
      */
-    private function resume($resumeToken = null)
+    private function resume($resumeToken = null, $hasAdvanced = false)
     {
         if (isset($resumeToken) && ! is_array($resumeToken) && ! is_object($resumeToken)) {
             throw InvalidArgumentException::invalidType('$resumeToken', $resumeToken, 'array or object');
@@ -347,12 +348,14 @@ private function resume($resumeToken = null)
         // Select a new server using the original read preference
         $server = $this->manager->selectServer($this->aggregateOptions['readPreference']);
 
+        $resumeOption = isset($this->changeStreamOptions['startAfter']) && !$hasAdvanced ? 'startAfter' : 'resumeAfter';
+
         unset($this->changeStreamOptions['resumeAfter']);
         unset($this->changeStreamOptions['startAfter']);
         unset($this->changeStreamOptions['startAtOperationTime']);
 
         if ($resumeToken !== null) {
-            $this->changeStreamOptions['resumeAfter'] = $resumeToken;
+            $this->changeStreamOptions[$resumeOption] = $resumeToken;
         }
 
         if ($resumeToken === null && $this->operationTime !== null) {

tests/Operation/WatchFunctionalTest.php

@@ -1382,6 +1382,105 @@ public function testGetResumeTokenReturnsOriginalResumeTokenOnEmptyBatch()
         $this->assertSame($resumeToken, $changeStream->getResumeToken());
     }
 
+    /**
+     * Prose test: "$changeStream stage for ChangeStream started with startAfter
+     * against a server >=4.1.1 that has not received any results yet MUST
+     * include a startAfter option and MUST NOT include a resumeAfter option
+     * when resuming a change stream."
+     */
+    public function testResumingChangeStreamWithoutPreviousResultsIncludesStartAfterOption()
+    {
+        if (version_compare($this->getServerVersion(), '4.1.1', '<')) {
+            $this->markTestSkipped('Testing resumeAfter and startAfter can only be tested on servers >= 4.1.1');
+        }
+
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
+        $changeStream = $operation->execute($this->getPrimaryServer());
+
+        $this->insertDocument(['x' => 1]);
+
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $resumeToken = $changeStream->getResumeToken();
+
+        $options = ['startAfter' => $resumeToken] + $this->defaultOptions;
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $options);
+        $changeStream = $operation->execute($this->getPrimaryServer());
+        $changeStream->rewind();
+        $this->killChangeStreamCursor($changeStream);
+
+        $aggregateCommand = null;
+
+        (new CommandObserver)->observe(
+            function() use ($changeStream) {
+                $changeStream->next();
+            },
+            function(array $event) use (&$aggregateCommand) {
+                if ($event['started']->getCommandName() !== 'aggregate') {
+                    return;
+                }
+
+                $aggregateCommand = $event['started']->getCommand();
+            }
+        );
+
+        $this->assertNotNull($aggregateCommand);
+        $this->assertObjectNotHasAttribute('resumeAfter', $aggregateCommand->pipeline[0]->{'$changeStream'});
+        $this->assertObjectHasAttribute('startAfter', $aggregateCommand->pipeline[0]->{'$changeStream'});
+    }
+
+    /**
+     * Prose test: "$changeStream stage for ChangeStream started with startAfter
+     * against a server >=4.1.1 that has received at least one result MUST
+     * include a resumeAfter option and MUST NOT include a startAfter option
+     * when resuming a change stream."
+     */
+    public function testResumingChangeStreamWithPreviousResultsIncludesResumeAfterOption()
+    {
+        if (version_compare($this->getServerVersion(), '4.1.1', '<')) {
+            $this->markTestSkipped('Testing resumeAfter and startAfter can only be tested on servers >= 4.1.1');
+        }
+
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
+        $changeStream = $operation->execute($this->getPrimaryServer());
+
+        $this->insertDocument(['x' => 1]);
+
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $resumeToken = $changeStream->getResumeToken();
+
+        $options = ['startAfter' => $resumeToken] + $this->defaultOptions;
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $options);
+        $changeStream = $operation->execute($this->getPrimaryServer());
+        $changeStream->rewind();
+
+        $this->insertDocument(['x' => 2]);
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+
+        $this->killChangeStreamCursor($changeStream);
+
+        $aggregateCommand = null;
+
+        (new CommandObserver)->observe(
+            function() use ($changeStream) {
+                $changeStream->next();
+            },
+            function(array $event) use (&$aggregateCommand) {
+                if ($event['started']->getCommandName() !== 'aggregate') {
+                    return;
+                }
+
+                $aggregateCommand = $event['started']->getCommand();
+            }
+        );
+
+        $this->assertNotNull($aggregateCommand);
+        $this->assertObjectNotHasAttribute('startAfter', $aggregateCommand->pipeline[0]->{'$changeStream'});
+        $this->assertObjectHasAttribute('resumeAfter', $aggregateCommand->pipeline[0]->{'$changeStream'});
+    }
+
     private function assertNoCommandExecuted(callable $callable)
     {
         $commands = [];