PHPLIB-456: Always invalidate ChangeStream position if iteration fails

Author: alcaeus

src/Model/ChangeStreamIterator.php

@@ -48,6 +48,7 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
     private $batchPosition = 0;
     private $batchSize;
     private $isRewindNop;
+    private $isValid = false;
     private $postBatchResumeToken;
     private $resumeToken;
 
@@ -117,6 +118,15 @@ final public function commandSucceeded(CommandSucceededEvent $event)
         }
     }
 
+    /**
+     * @see https://php.net/iteratoriterator.current
+     * @return mixed
+     */
+    public function current()
+    {
+        return $this->isValid ? parent::current() : null;
+    }
+
     /**
      * Returns the resume token for the iterator's current position.
      *
@@ -131,6 +141,15 @@ public function getResumeToken()
         return $this->resumeToken;
     }
 
+    /**
+     * @see https://php.net/iteratoriterator.key
+     * @return mixed
+     */
+    public function key()
+    {
+        return $this->isValid ? parent::key() : null;
+    }
+
     /**
      * @see https://php.net/iteratoriterator.rewind
      * @return void
@@ -172,6 +191,15 @@ public function rewind()
         $this->onIteration(false);
     }
 
+    /**
+     * @see https://php.net/iteratoriterator.valid
+     * @return boolean
+     */
+    public function valid()
+    {
+        return $this->isValid;
+    }
+
     /**
      * Extracts the resume token (i.e. "_id" field) from a change document.
      *
@@ -195,10 +223,12 @@ private function extractResumeToken($document)
             : (isset($document->_id) ? $document->_id : null);
 
         if (! isset($resumeToken)) {
+            $this->isValid = false;
             throw ResumeTokenException::notFound();
         }
 
         if (! is_array($resumeToken) && ! is_object($resumeToken)) {
+            $this->isValid = false;
             throw ResumeTokenException::invalidType($resumeToken);
         }
 
@@ -223,16 +253,16 @@ private function isAtEndOfBatch()
      */
     private function onIteration($incrementBatchPosition)
     {
-        $isValid = $this->valid();
+        $this->isValid = parent::valid();
 
         /* Disable rewind()'s NOP behavior once we advance to a valid position.
          * This will allow the driver to throw a LogicException if rewind() is
          * called after the cursor has advanced past its first element. */
-        if ($this->isRewindNop && $isValid) {
+        if ($this->isRewindNop && $this->isValid) {
             $this->isRewindNop = false;
         }
 
-        if ($incrementBatchPosition && $isValid) {
+        if ($incrementBatchPosition && $this->isValid) {
             $this->batchPosition++;
         }
 
@@ -244,7 +274,7 @@ private function onIteration($incrementBatchPosition)
          * from the current document if possible. */
         if ($this->isAtEndOfBatch() && $this->postBatchResumeToken !== null) {
             $this->resumeToken = $this->postBatchResumeToken;
-        } elseif ($isValid) {
+        } elseif ($this->isValid) {
             $this->resumeToken = $this->extractResumeToken($this->current());
         }
     }

tests/Operation/WatchFunctionalTest.php

@@ -1123,34 +1123,21 @@ public function testResumeTokenNotFoundDoesNotAdvanceKey()
             $changeStream->next();
             $this->fail('Exception for missing resume token was not thrown');
         } catch (ResumeTokenException $e) {
-            /* If a client-side error is thrown (server < 4.1.8), the tailable
-             * cursor's position is still valid. This may change once PHPLIB-456
-             * is implemented. */
-            $expectedValid = true;
-            $expectedKey = 0;
         } catch (ServerException $e) {
-            /* If a server-side error is thrown (server >= 4.1.8), the tailable
-             * cursor's position is not valid. */
-            $expectedValid = false;
-            $expectedKey = null;
         }
 
-        $this->assertSame($expectedValid, $changeStream->valid());
-        $this->assertSame($expectedKey, $changeStream->key());
+        $this->assertFalse($changeStream->valid());
+        $this->assertNull($changeStream->key());
 
         try {
             $changeStream->next();
             $this->fail('Exception for missing resume token was not thrown');
         } catch (ResumeTokenException $e) {
-            $expectedValid = true;
-            $expectedKey = 0;
         } catch (ServerException $e) {
-            $expectedValid = false;
-            $expectedKey = null;
         }
 
-        $this->assertSame($expectedValid, $changeStream->valid());
-        $this->assertSame($expectedKey, $changeStream->key());
+        $this->assertFalse($changeStream->valid());
+        $this->assertNull($changeStream->key());
     }
 
     public function testSessionPersistsAfterResume()