PHPLIB-276: Add maxAwaitTimeMS support for change streams

Author: kvwalker

src/Operation/Aggregate.php

@@ -163,6 +163,10 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
             throw InvalidArgumentException::invalidType('"hint" option', $options['hint'], 'string or array or object');
         }
 
+        if (isset($options['maxAwaitTimeMS']) && ! is_integer($options['maxAwaitTimeMS'])) {
+            throw InvalidArgumentException::invalidType('"maxAwaitTimeMS" option', $options['maxAwaitTimeMS'], 'integer');
+        }
+
         if (isset($options['maxTimeMS']) && ! is_integer($options['maxTimeMS'])) {
             throw InvalidArgumentException::invalidType('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
         }
@@ -277,6 +281,7 @@ private function createCommand(Server $server, $isCursorSupported)
             'aggregate' => $this->collectionName,
             'pipeline' => $this->pipeline,
         ];
+        $cmdOptions = [];
 
         // Servers < 2.6 do not support any command options
         if ( ! $isCursorSupported) {
@@ -303,13 +308,17 @@ private function createCommand(Server $server, $isCursorSupported)
             $cmd['hint'] = is_array($this->options['hint']) ? (object) $this->options['hint'] : $this->options['hint'];
         }
 
+        if (isset($this->options['maxAwaitTimeMS'])) {
+            $cmdOptions['maxAwaitTimeMS'] = $this->options['maxAwaitTimeMS'];
+        }
+
         if ($this->options['useCursor']) {
             $cmd['cursor'] = isset($this->options["batchSize"])
                 ? ['batchSize' => $this->options["batchSize"]]
                 : new stdClass;
         }
 
-        return new Command($cmd);
+        return new Command($cmd, $cmdOptions);
     }
 
     /**

src/Operation/ChangeStream.php

@@ -147,7 +147,7 @@ public function execute(Server $server)
 
     private function createAggregateOptions()
     {
-        $aggOptions = array_intersect_key($this->options, ['batchSize' => 1, 'collation' => 1]);
+        $aggOptions = array_intersect_key($this->options, ['batchSize' => 1, 'collation' => 1, 'maxAwaitTimeMS' => 1]);
         if ( ! $aggOptions) {
             return [];
         }
@@ -174,6 +174,7 @@ private function createCommand()
         array_unshift($this->pipeline, $changeStreamArray);
 
         $cmd = new Aggregate($this->databaseName, $this->collectionName, $this->pipeline, $this->createAggregateOptions());
+
         return $cmd;
     }
 

tests/Operation/ChangeStreamFunctionalTest.php

@@ -237,4 +237,21 @@ public function testConnectionException()
                         ]);
         $this->assertEquals($changeStreamResult->current(), $expectedResult);
     }
+
+    public function testMaxAwaitTimeMS()
+    {
+        $this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
+        $maxAwaitTimeMS = 10;
+        $changeStreamResult = $this->collection->watch([], ['maxAwaitTimeMS' => $maxAwaitTimeMS]);
+
+        /* Make sure we await results for at least maxAwaitTimeMS, since no new
+         * documents should be inserted to wake up the server's command thread.
+         * Also ensure that we don't wait too long (server default is one
+         * second). */
+        $startTime = microtime(true);
+        $changeStreamResult->rewind();
+        $this->assertGreaterThanOrEqual($maxAwaitTimeMS * 0.001, microtime(true) - $startTime);
+        $this->assertLessThan(0.5, microtime(true) - $startTime);
+   }
+
 }

tests/Operation/FindFunctionalTest.php

@@ -154,7 +154,10 @@ public function testMaxAwaitTimeMS()
         $cursor = $operation->execute($this->getPrimaryServer());
         $it = new \IteratorIterator($cursor);
 
-        // Make sure we await results for no more than the maxAwaitTimeMS.
+        /* Make sure we await results for at least maxAwaitTimeMS, since no new
+         * documents should be inserted to wake up the server's query thread.
+         * Also ensure that we don't wait too long (server default is one
+         * second). */
         $it->rewind();
         $it->next();
         $startTime = microtime(true);