Add testSessionFreed and fix cursorId check in ChangeStream::next()

Author: kvwalker

src/ChangeStream.php

@@ -107,8 +107,8 @@ public function next()
              * free any reference to Watch. This will also free the only
              * reference to an implicit session, since any such reference
              * belongs to Watch. */
-            if ($this->getCursorId() === 0) {
-                unset($this->resumeCallable);
+            if ((string)$this->getCursorId() === '0') {
+                $this->resumeCallable = null;
             }
         } catch (RuntimeException $e) {
             if (strpos($e->getMessage(), "not master") !== false) {
@@ -140,8 +140,8 @@ public function rewind()
                 $this->resumeToken = $this->extractResumeToken($this->csIt->current());
             }
             // As with next(), free the callable once we know it will never be used.
-            if ($this->getCursorId() === 0) {
-                unset($this->resumeCallable);
+            if ((string)$this->getCursorId() === '0') {
+                $this->resumeCallable = null;
             }
         } catch (RuntimeException $e) {
             if (strpos($e->getMessage(), "not master") !== false) {

tests/Operation/WatchFunctionalTest.php

@@ -8,7 +8,9 @@
 use MongoDB\Driver\Server;
 use MongoDB\Driver\Exception\ConnectionTimeoutException;
 use MongoDB\Exception\ResumeTokenException;
+use MongoDB\Operation\CreateCollection;
 use MongoDB\Operation\DatabaseCommand;
+use MongoDB\Operation\DropCollection;
 use MongoDB\Operation\InsertOne;
 use MongoDB\Operation\Watch;
 use MongoDB\Tests\CommandObserver;
@@ -627,16 +629,10 @@ function ($changeStream) use (&$sessionAfterResume, &$commands) {
         );
 
         $expectedCommands = [
-            /* The initial aggregate command for change streams returns a cursor
-             * envelope with an empty initial batch, since there are no changes
-             * to report at the moment the change stream is created. Therefore,
-             * we expect a getMore to be issued when we first advance the change
-             * stream (with either rewind() or next()). */
+            /* We expect a getMore to be issued because we are calling next(). */
             'getMore',
-            /* Since socketTimeoutMS is less than maxAwaitTimeMS, the previous
-             * getMore command encounters a client socket timeout and leaves the
-             * cursor open on the server. ChangeStream should catch this error
-             * and resume by issuing a new aggregate command. */
+            /* Since we have killed the cursor, ChangeStream will resume by
+             * issuing a new aggregate commmand. */
             'aggregate',
             /* When ChangeStream resumes, it overwrites its original cursor with
              * the new cursor resulting from the last aggregate command. This
@@ -653,7 +649,29 @@ function ($changeStream) use (&$sessionAfterResume, &$commands) {
         foreach ($sessionAfterResume as $session) {
             $this->assertEquals($session, $originalSession);
         }
+    }
+
+    public function testSessionFreed()
+    {
+        $operation = new CreateCollection($this->getDatabaseName(), $this->getCollectionName());
+        $operation->execute($this->getPrimaryServer());
+
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
+        $changeStream = $operation->execute($this->getPrimaryServer());
+
+        $rc = new ReflectionClass($changeStream);
+        $rp = $rc->getProperty('resumeCallable');
+        $rp->setAccessible(true);
+
+        $this->assertNotNull($rp->getValue($changeStream));
+
+        // Invalidate the cursor to verify that resumeCallable is unset when the cursor is exhausted.
+        $operation = new DropCollection($this->getDatabaseName(), $this->getCollectionName());
+        $operation->execute($this->getPrimaryServer());
+
+        $changeStream->next();
 
+        $this->assertNull($rp->getValue($changeStream));
     }
 
     private function insertDocument($document)