PHPLIB-342: Change streams should use the same session when resuming

Author: kvwalker

src/ChangeStream.php

@@ -101,6 +101,15 @@ public function next()
                 $this->hasAdvanced = true;
                 $this->resumeToken = $this->extractResumeToken($this->csIt->current());
             }
+            // If the cursorId is 0, the server has invalidated the cursor so we
+            // will never perform another getMore. This means that we cannot
+            // resume and we can therefore unset the resumeCallable, which will
+            // 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);
+            }
         } catch (RuntimeException $e) {
             if (strpos($e->getMessage(), "not master") !== false) {
                 $resumable = true;
@@ -130,6 +139,15 @@ public function rewind()
                 $this->hasAdvanced = true;
                 $this->resumeToken = $this->extractResumeToken($this->csIt->current());
             }
+            // If the cursorId is 0, the server has invalidated the cursor so we
+            // will never perform another getMore. This means that we cannot
+            // resume and we can therefore unset the resumeCallable, which will
+            // 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);
+            }
         } catch (RuntimeException $e) {
             if (strpos($e->getMessage(), "not master") !== false) {
                 $resumable = true;

src/Operation/Watch.php

@@ -110,6 +110,12 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
             }
         }
 
+        if ( ! isset($options['session'])) {
+            try {
+                $options['session'] = $manager->startSession();
+            } catch (DriverRuntimeException $e) {}
+        }
+
         $this->databaseName = (string) $databaseName;
         $this->collectionName = (string) $collectionName;
         $this->pipeline = $pipeline;

tests/Operation/WatchFunctionalTest.php

@@ -588,6 +588,46 @@ public function testResumeTokenNotFoundAdvancesKey()
         $this->assertSame(2, $changeStream->key());
     }
 
+    public function testSessionPersistsAfterResume()
+    {
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
+
+        $changeStream = null;
+        $originalSession = null;
+        $sessionAfterResume = null;
+
+        // We want to ensure that the lsid of the initial aggregate matches the
+        // lsid of any aggregates after the change stream resumes. After
+        // PHPC-1152 is complete, we will ensure that the lsid of the initial
+        // aggregate matches the lsid of any subsequent aggregates and getMores.
+        (new CommandObserver)->observe(
+            function() use ($operation, &$changeStream) {
+                $changeStream = $operation->execute($this->getPrimaryServer());
+            },
+            function($changeStream) use (&$originalSession) {
+                if (isset($changeStream->aggregate)) {
+                    $originalSession = $changeStream->lsid;
+                }
+            }
+        );
+
+        $changeStream->rewind();
+        $this->killChangeStreamCursor($changeStream);
+
+        (new CommandObserver)->observe(
+            function() use (&$changeStream) {
+                $changeStream->next();
+            },
+            function ($changeStream) use (&$sessionAfterResume) {
+                if (isset($changeStream->aggregate)) {
+                    $sessionAfterResume = $changeStream->lsid;
+                }
+            }
+        );
+
+        $this->assertEquals($sessionAfterResume, $originalSession);
+    }
+
     private function insertDocument($document)
     {
         $insertOne = new InsertOne($this->getDatabaseName(), $this->getCollectionName(), $document);