PHPLIB-443: Propagate non-resumable errors during ChangeStream iteration

src/ChangeStream.php

@@ -120,9 +120,7 @@ class ChangeStream implements Iterator
                 $this->resumeCallable = null;
             }
         } catch (RuntimeException $e) {
-            if ($this->isResumableError($e)) {
-                $this->resume();
-            }
+            $this->resumeOrThrow($e);
         }
     }
 
@@ -144,9 +142,7 @@ class ChangeStream implements Iterator
                 $this->resumeCallable = null;
             }
         } catch (RuntimeException $e) {
-            if ($this->isResumableError($e)) {
-                $this->resume();
-            }
+            $this->resumeOrThrow($e);
         }
     }
 
@@ -227,4 +223,20 @@ class ChangeStream implements Iterator
         $this->csIt = $newChangeStream->csIt;
         $this->csIt->rewind();
     }
+
+    /**
+     * Either resumes after a resumable error or re-throws the exception.
+     *
+     * @param RuntimeException $exception
+     * @throws RuntimeException
+     */
+    private function resumeOrThrow(RuntimeException $exception)
+    {
+        if ($this->isResumableError($exception)) {
+            $this->resume();
+            return;
+        }
+
+        throw $exception;
+    }
 }

tests/SpecTests/ChangeStreamsProseTest.php

+<?php
+
+namespace MongoDB\Tests\SpecTests;
+
+use MongoDB\Collection;
+use MongoDB\Driver\Exception\ServerException;
+
+/**
+ * Change Streams spec prose tests.
+ *
+ * @see https://github.com/mongodb/specifications/tree/master/source/change-streams
+ */
+class ChangeStreamsProseTest extends FunctionalTestCase
+{
+    private $collection;
+
+    public function setUp()
+    {
+        parent::setUp();
+
+        $this->skipIfChangeStreamIsNotSupported();
+
+        $this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
+        $this->dropCollection();
+    }
+
+    public function tearDown()
+    {
+        if (!$this->hasFailed()) {
+            $this->dropCollection();
+        }
+
+        parent::tearDown();
+    }
+
+    /**
+     * ChangeStream will not attempt to resume after encountering error code
+     * 11601 (Interrupted), 136 (CappedPositionLost), or 237 (CursorKilled)
+     * while executing a getMore command.
+     *
+     * @dataProvider provideNonResumableErrorCodes
+     */
+    public function testProseTest5($errorCode)
+    {
+        $this->configureFailPoint([
+            'configureFailPoint' => 'failCommand',
+            'mode' => ['times' => 1],
+            'data' => ['failCommands' => ['getMore'], 'errorCode' => $errorCode],
+        ]);
+
+        $this->createCollection();
+        $changeStream = $this->collection->watch();
+
+        $this->expectException(ServerException::class);
+        $this->expectExceptionCode($errorCode);
+        $changeStream->rewind();
+    }
+
+    public function provideNonResumableErrorCodes()
+    {
+        return [
+            [136], // CappedPositionLost
+            [237], // CursorKilled
+            [11601], // Interrupted
+        ];
+    }
+}

tests/SpecTests/FunctionalTestCase.php

@@ -147,11 +147,19 @@ class FunctionalTestCase extends BaseFunctionalTestCase
      * The fail point will automatically be disabled during tearDown() to avoid
      * affecting a subsequent test.
      *
-     * @param stdClass $command configureFailPoint command document
+     * @param array|stdClass $command configureFailPoint command document
      * @throws InvalidArgumentException if $command is not a configureFailPoint command
      */
-    protected function configureFailPoint(stdClass $command)
+    protected function configureFailPoint($command)
     {
+        if (is_array($command)) {
+            $command = (object) $command;
+        }
+
+        if ( ! $command instanceof stdClass) {
+            throw new InvalidArgumentException('$command is not an array or stdClass instance');
+        }
+
         if (key($command) !== 'configureFailPoint') {
             throw new InvalidArgumentException('$command is not a configureFailPoint command');
         }