Test expectedCommands to ensure killChangeStreamCursor doesn't need to take implicit session

Author: kvwalker

tests/Operation/WatchFunctionalTest.php

@@ -594,12 +594,14 @@ public function testSessionPersistsAfterResume()
 
         $changeStream = null;
         $originalSession = null;
-        $sessionAfterResume = null;
+        $sessionAfterResume = [];
+        $commands = [];
 
-        // 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.
+        /* 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());
@@ -618,14 +620,40 @@ function($changeStream) use (&$originalSession) {
             function() use (&$changeStream) {
                 $changeStream->next();
             },
-            function ($changeStream) use (&$sessionAfterResume) {
-                if (isset($changeStream->aggregate)) {
-                    $sessionAfterResume = bin2hex((string) $changeStream->lsid->id);
-                }
+            function ($changeStream) use (&$sessionAfterResume, &$commands) {
+                $commands[] = key((array) $changeStream);
+                $sessionAfterResume[] = bin2hex((string) $changeStream->lsid->id);
             }
         );
 
-        $this->assertEquals($sessionAfterResume, $originalSession);
+        $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()). */
+            '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. */
+            'aggregate',
+            /* When ChangeStream resumes, it overwrites its original cursor with
+             * the new cursor resulting from the last aggregate command. This
+             * removes the last reference to the old cursor, which causes the
+             * driver to kill it (via mongoc_cursor_destroy()). */
+            'killCursors',
+            /* Finally, ChangeStream will rewind the new cursor as the last step
+             * of the resume process. This results in one last getMore. */
+            'getMore',
+        ];
+
+        $this->assertSame($expectedCommands, $commands);
+
+        foreach ($sessionAfterResume as $session) {
+            $this->assertEquals($session, $originalSession);
+        }
+
     }
 
     private function insertDocument($document)