Merge pull request #468

Author: kvwalker

docs/includes/apiargs-MongoDBCollection-method-watch-option.yaml

@@ -22,9 +22,15 @@ interface: phpmethod
 operation: ~
 optional: true
 ---
-source:
-  file: apiargs-MongoDBCollection-method-find-option.yaml
-  ref: maxAwaitTimeMS
+arg_name: option
+name: maxAwaitTimeMS
+type: integer
+description: |
+  Positive integer denoting the time limit in milliseconds for the server to
+  block a getMore operation if no data is available.
+interface: phpmethod
+operation: ~
+optional: true
 ---
 source:
   file: apiargs-MongoDBCollection-common-option.yaml

docs/tutorial/tailable-cursor.txt

@@ -177,7 +177,6 @@ using :php:`IteratorIterator <iteratoriterator>`.
 
       $iterator->next();
    }
-}
 
 Much like the ``foreach`` example, this version on the consumer script will
 start by quickly printing all results in the capped collection; however, it will

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,49 @@ public function testConnectionException()
                         ]);
         $this->assertEquals($changeStreamResult->current(), $expectedResult);
     }
+
+    public function testMaxAwaitTimeMS()
+    {
+        $this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
+        /* On average, an acknowledged write takes about 20 ms to appear in a
+         * change stream on the server so we'll use a higher maxAwaitTimeMS to
+         * ensure we see the write. */
+        $maxAwaitTimeMS = 100;
+        $changeStreamResult = $this->collection->watch([], ['maxAwaitTimeMS' => $maxAwaitTimeMS]);
+
+        /* The initial change stream is empty so we should expect a delay when
+         * we call rewind, since it issues a getMore. Expect to wait at least
+         * maxAwaitTimeMS, since no new documents should be inserted to wake up
+         * the server's query thread. Also ensure we don't wait too long (server
+         * default is one second). */
+        $startTime = microtime(true);
+        $changeStreamResult->rewind();
+        $duration = microtime(true) - $startTime;
+        $this->assertGreaterThanOrEqual($maxAwaitTimeMS * 0.001, $duration);
+        $this->assertLessThan(0.5, $duration);
+
+        $this->assertFalse($changeStreamResult->valid());
+
+        /* Advancing again on a change stream will issue a getMore, so we should
+         * expect a delay again. */
+        $startTime = microtime(true);
+        $changeStreamResult->next();
+        $duration = microtime(true) - $startTime;
+        $this->assertGreaterThanOrEqual($maxAwaitTimeMS * 0.001, $duration);
+        $this->assertLessThan(0.5, $duration);
+
+        $this->assertFalse($changeStreamResult->valid());
+
+        /* After inserting a document, the change stream will not issue a
+         * getMore so we should not expect a delay. */
+        $result = $this->collection->insertOne(['_id' => 1]);
+        $this->assertInstanceOf('MongoDB\InsertOneResult', $result);
+        $this->assertSame(1, $result->getInsertedCount());
+
+        $startTime = microtime(true);
+        $changeStreamResult->next();
+        $duration = microtime(true) - $startTime;
+        $this->assertLessThan($maxAwaitTimeMS * 0.001, $duration);
+        $this->assertTrue($changeStreamResult->valid());
+    }
 }

tests/Operation/FindFunctionalTest.php

@@ -148,18 +148,45 @@ public function testMaxAwaitTimeMS()
         // Insert documents into the capped collection.
         $bulkWrite = new BulkWrite(['ordered' => true]);
         $bulkWrite->insert(['_id' => 1]);
+        $bulkWrite->insert(['_id' => 2]);
         $result = $this->manager->executeBulkWrite($this->getNamespace(), $bulkWrite);
 
         $operation = new Find($databaseName, $cappedCollectionName, [], ['cursorType' => Find::TAILABLE_AWAIT, 'maxAwaitTimeMS' => $maxAwaitTimeMS]);
         $cursor = $operation->execute($this->getPrimaryServer());
         $it = new \IteratorIterator($cursor);
 
-        // Make sure we await results for no more than the maxAwaitTimeMS.
+        /* The initial query includes the one and only document in its result
+         * batch, so we should not expect a delay. */
+        $startTime = microtime(true);
         $it->rewind();
+        $duration = microtime(true) - $startTime;
+        $this->assertLessThan($maxAwaitTimeMS * 0.001, $duration);
+
+        $this->assertTrue($it->valid());
+        $this->assertSameDocument(['_id' => 1], $it->current());
+
+        /* Advancing again takes us to the last document of the result batch,
+         * but still should not issue a getMore */
+        $startTime = microtime(true);
         $it->next();
+        $duration = microtime(true) - $startTime;
+        $this->assertLessThan($maxAwaitTimeMS * 0.001, $duration);
+
+        $this->assertTrue($it->valid());
+        $this->assertSameDocument(['_id' => 2], $it->current());
+
+        /* Now that we've reached the end of the initial result batch, advancing
+         * again will issue a getMore. Expect to wait at least maxAwaitTimeMS,
+         * since no new documents should be inserted to wake up the server's
+         * query thread. Also ensure we don't wait too long (server default is
+         * one second). */
         $startTime = microtime(true);
         $it->next();
-        $this->assertGreaterThanOrEqual($maxAwaitTimeMS * 0.001, microtime(true) - $startTime);
+        $duration = microtime(true) - $startTime;
+        $this->assertGreaterThanOrEqual($maxAwaitTimeMS * 0.001, $duration);
+        $this->assertLessThan(0.5, $duration);
+
+        $this->assertFalse($it->valid());
     }
 
     /**