Merge pull request #636

src/ChangeStream.php

@@ -24,7 +24,7 @@ use MongoDB\Driver\Exception\RuntimeException;
 use MongoDB\Driver\Exception\ServerException;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\ResumeTokenException;
-use IteratorIterator;
+use MongoDB\Model\TailableCursorIterator;
 use Iterator;
 
 /**
@@ -63,11 +63,12 @@ class ChangeStream implements Iterator
      * @internal
      * @param Cursor   $cursor
      * @param callable $resumeCallable
+     * @param boolean  $isFirstBatchEmpty
      */
-    public function __construct(Cursor $cursor, callable $resumeCallable)
+    public function __construct(Cursor $cursor, callable $resumeCallable, $isFirstBatchEmpty)
     {
         $this->resumeCallable = $resumeCallable;
-        $this->csIt = new IteratorIterator($cursor);
+        $this->csIt = new TailableCursorIterator($cursor, $isFirstBatchEmpty);
     }
 
     /**
@@ -242,17 +243,11 @@ class ChangeStream implements Iterator
      */
     private function resume()
     {
-        $newChangeStream = call_user_func($this->resumeCallable, $this->resumeToken);
-        $this->csIt = $newChangeStream->csIt;
+        list($cursor, $isFirstBatchEmpty) = call_user_func($this->resumeCallable, $this->resumeToken);
+
+        $this->csIt = new TailableCursorIterator($cursor, $isFirstBatchEmpty);
         $this->csIt->rewind();
-        /* Note: if we are resuming after a call to ChangeStream::rewind(),
-         * $hasAdvanced will always be false. For it to be true, rewind() would
-         * need to have thrown a RuntimeException with a resumable error, which
-         * can only happen during the first call to IteratorIterator::rewind()
-         * before onIteration() has a chance to set $hasAdvanced to true.
-         * Otherwise, IteratorIterator::rewind() would either NOP (consecutive
-         * rewinds) or throw a LogicException (rewind after next), neither of
-         * which would result in a call to resume(). */
+
         $this->onIteration($this->hasAdvanced);
     }
 

src/Model/TailableCursorIterator.php

+<?php
+/*
+ * Copyright 2019 MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace MongoDB\Model;
+
+use MongoDB\Driver\Cursor;
+use IteratorIterator;
+
+/**
+ * Iterator for tailable cursors.
+ *
+ * This iterator may be used to wrap a tailable cursor. By indicating whether
+ * the cursor's first batch of results is empty, this iterator can NOP initial
+ * calls to rewind() and prevent it from executing a getMore command.
+ *
+ * @internal
+ */
+class TailableCursorIterator extends IteratorIterator
+{
+    private $isRewindNop;
+
+    /**
+     * Constructor.
+     *
+     * @internal
+     * @param Cursor  $cursor
+     * @param boolean $isFirstBatchEmpty
+     */
+    public function __construct(Cursor $cursor, $isFirstBatchEmpty)
+    {
+        parent::__construct($cursor);
+        $this->isRewindNop = $isFirstBatchEmpty;
+    }
+
+    /**
+     * @see https://php.net/iteratoriterator.rewind
+     * @return void
+     */
+    public function next()
+    {
+        try {
+            parent::next();
+        } finally {
+            /* If the cursor ever advances to a valid position, do not prevent
+             * future attempts to rewind the cursor. This will allow the driver
+             * to throw a LogicException if the cursor has been advanced past
+             * its first element. */
+            if ($this->valid()) {
+                $this->isRewindNop = false;
+            }
+        }
+    }
+
+    /**
+     * @see https://php.net/iteratoriterator.rewind
+     * @return void
+     */
+    public function rewind()
+    {
+        if ($this->isRewindNop) {
+            return;
+        }
+
+        parent::rewind();
+    }
+}

src/Operation/Watch.php

@@ -57,6 +57,7 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
     private $changeStreamOptions;
     private $collectionName;
     private $databaseName;
+    private $isFirstBatchEmpty = false;
     private $operationTime;
     private $pipeline;
     private $resumeCallable;
@@ -200,6 +201,11 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
     /** @internal */
     final public function commandStarted(CommandStartedEvent $event)
     {
+        if ($event->getCommandName() !== 'aggregate') {
+            return;
+        }
+
+        $this->isFirstBatchEmpty = false;
     }
 
     /** @internal */
@@ -211,9 +217,15 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
 
         $reply = $event->getReply();
 
-        if (isset($reply->operationTime) && $reply->operationTime instanceof TimestampInterface) {
+        /* Note: the spec only refers to collecting an operation time from the
+         * "original aggregation", so only capture it if we've not already. */
+        if (!isset($this->operationTime) && isset($reply->operationTime) && $reply->operationTime instanceof TimestampInterface) {
             $this->operationTime = $reply->operationTime;
         }
+
+        if (isset($reply->cursor->firstBatch) && is_array($reply->cursor->firstBatch)) {
+            $this->isFirstBatchEmpty = empty($reply->cursor->firstBatch);
+        }
     }
 
     /**
@@ -227,7 +239,9 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
      */
     public function execute(Server $server)
     {
-        return new ChangeStream($this->executeAggregate($server), $this->resumeCallable);
+        $cursor = $this->executeAggregate($server);
+
+        return new ChangeStream($cursor, $this->resumeCallable, $this->isFirstBatchEmpty);
     }
 
     /**
@@ -255,40 +269,36 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
                 unset($this->changeStreamOptions['startAtOperationTime']);
             }
 
+            // Select a new server using the original read preference
+            $server = $manager->selectServer($this->aggregateOptions['readPreference']);
+
             /* If we captured an operation time from the first aggregate command
              * and there is no "resumeAfter" option, set "startAtOperationTime"
              * so that we can resume from the original aggregate's time. */
-            if ($this->operationTime !== null && ! isset($this->changeStreamOptions['resumeAfter'])) {
+            if ($this->operationTime !== null && ! isset($this->changeStreamOptions['resumeAfter']) &&
+                \MongoDB\server_supports_feature($server, self::$wireVersionForStartAtOperationTime)) {
                 $this->changeStreamOptions['startAtOperationTime'] = $this->operationTime;
             }
 
+            // Recreate the aggregate command and execute to obtain a new cursor
             $this->aggregate = $this->createAggregate();
+            $cursor = $this->executeAggregate($server);
 
-            /* Select a new server using the read preference, execute this
-             * operation on it, and return the new ChangeStream. */
-            $server = $manager->selectServer($this->aggregateOptions['readPreference']);
-
-            return $this->execute($server);
+            return [$cursor, $this->isFirstBatchEmpty];
         };
     }
 
     /**
-     * Execute the aggregate command and optionally capture its operation time.
+     * Execute the aggregate command.
+     *
+     * The command will be executed using APM so that we can capture its
+     * operation time and/or firstBatch size.
      *
      * @param Server $server
      * @return Cursor
      */
     private function executeAggregate(Server $server)
     {
-        /* If we've already captured an operation time or the server does not
-         * support resuming from an operation time (e.g. MongoDB 3.6), execute
-         * the aggregation directly and return its cursor. */
-        if ($this->operationTime !== null || ! \MongoDB\server_supports_feature($server, self::$wireVersionForStartAtOperationTime)) {
-            return $this->aggregate->execute($server);
-        }
-
-        /* Otherwise, execute the aggregation using command monitoring so that
-         * we can capture its operation time with commandSucceeded(). */
         \MongoDB\Driver\Monitoring\addSubscriber($this);
 
         try {

tests/Model/TailableCursorIteratorTest.php

+<?php
+
+namespace MongoDB\Tests\Model;
+
+use MongoDB\Collection;
+use MongoDB\Driver\Exception\LogicException;
+use MongoDB\Model\TailableCursorIterator;
+use MongoDB\Operation\Find;
+use MongoDB\Operation\CreateCollection;
+use MongoDB\Operation\DropCollection;
+use MongoDB\Tests\CommandObserver;
+use MongoDB\Tests\FunctionalTestCase;
+
+class TailableCursorIteratorTest extends FunctionalTestCase
+{
+    private $collection;
+
+    public function setUp()
+    {
+        parent::setUp();
+
+        $operation = new DropCollection($this->getDatabaseName(), $this->getCollectionName());
+        $operation->execute($this->getPrimaryServer());
+
+        $operation = new CreateCollection($this->getDatabaseName(), $this->getCollectionName(), ['capped' => true, 'size' => 8192]);
+        $operation->execute($this->getPrimaryServer());
+
+        $this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
+    }
+
+    public function testFirstBatchIsEmpty()
+    {
+        $this->collection->insertOne(['x' => 1]);
+
+        $cursor = $this->collection->find(['x' => ['$gt' => 1]], ['cursorType' => Find::TAILABLE]);
+        $iterator = new TailableCursorIterator($cursor, true);
+
+        $this->assertNoCommandExecuted(function() use ($iterator) { $iterator->rewind(); });
+        $this->assertFalse($iterator->valid());
+
+        $this->collection->insertOne(['x' => 2]);
+
+        $iterator->next();
+        $this->assertTrue($iterator->valid());
+        $this->assertMatchesDocument(['x' => 2], $iterator->current());
+
+        $this->expectException(LogicException::class);
+        $iterator->rewind();
+    }
+
+    public function testFirstBatchIsNotEmpty()
+    {
+        $this->collection->insertOne(['x' => 1]);
+
+        $cursor = $this->collection->find([], ['cursorType' => Find::TAILABLE]);
+        $iterator = new TailableCursorIterator($cursor, false);
+
+        $this->assertNoCommandExecuted(function() use ($iterator) { $iterator->rewind(); });
+        $this->assertTrue($iterator->valid());
+        $this->assertMatchesDocument(['x' => 1], $iterator->current());
+
+        $this->collection->insertOne(['x' => 2]);
+
+        $iterator->next();
+        $this->assertTrue($iterator->valid());
+        $this->assertMatchesDocument(['x' => 2], $iterator->current());
+
+        $this->expectException(LogicException::class);
+        $iterator->rewind();
+    }
+
+    private function assertNoCommandExecuted(callable $callable)
+    {
+        $commands = [];
+
+        (new CommandObserver)->observe(
+            $callable,
+            function(array $event) use (&$commands) {
+                $this->fail(sprintf('"%s" command was executed', $event['started']->getCommandName()));
+            }
+        );
+
+        $this->assertEmpty($commands);
+    }
+}

tests/SpecTests/ChangeStreamsProseTest.php

@@ -54,10 +54,11 @@ class ChangeStreamsProseTest extends FunctionalTestCase
 
         $this->createCollection();
         $changeStream = $this->collection->watch();
+        $changeStream->rewind();
 
         $this->expectException(ServerException::class);
         $this->expectExceptionCode($errorCode);
-        $changeStream->rewind();
+        $changeStream->next();
     }
 
     public function provideNonResumableErrorCodes()

tests/SpecTests/ChangeStreamsSpecTest.php

@@ -233,22 +233,33 @@ class ChangeStreamsSpecTest extends FunctionalTestCase
      * Iterate a change stream.
      *
      * @param ChangeStream $changeStream
+     * @param integer      $limit
      * @return BSONDocument[]
      */
     private function iterateChangeStream(ChangeStream $changeStream, $limit = 0)
     {
+        if ($limit < 0) {
+            throw new LogicException('$limit is negative');
+        }
+
+        /* Limit iterations to guard against an infinite loop should a test fail
+         * to return as many results as are expected. Require at least one
+         * iteration to allow next() a chance to throw for error tests. */
+        $maxIterations = $limit + 1;
         $events = [];
 
-        for ($changeStream->rewind(); count($events) < $limit; $changeStream->next()) {
+        for ($i = 0, $changeStream->rewind(); $i < $maxIterations; $i++, $changeStream->next()) {
             if ( ! $changeStream->valid()) {
                 continue;
             }
 
             $event = $changeStream->current();
-
             $this->assertInstanceOf(BSONDocument::class, $event);
-
             $events[] = $event;
+
+            if (count($events) >= $limit) {
+                break;
+            }
         }
 
         return $events;

tests/SpecTests/Operation.php

@@ -6,6 +6,7 @@ use MongoDB\Collection;
 use MongoDB\Database;
 use MongoDB\Driver\Cursor;
 use MongoDB\Driver\Session;
+use MongoDB\Driver\WriteConcern;
 use MongoDB\Driver\Exception\BulkWriteException;
 use MongoDB\Driver\Exception\Exception;
 use MongoDB\Operation\FindOneAndReplace;
@@ -53,6 +54,11 @@ final class Operation
     {
         $o = new self($operation);
 
+        /* Note: change streams only return majority-committed writes, so ensure
+         * each operation applies that write concern. This will avoid spurious
+         * test failures. */
+        $writeConcern = new WriteConcern(WriteConcern::MAJORITY);
+
         // Expect all operations to succeed
         $o->errorExpectation = ErrorExpectation::noError();
 
@@ -66,6 +72,8 @@ final class Operation
             $o->arguments = ['command' => [
                 'renameCollection' => $operation->database . '.' . $operation->collection,
                 'to' => $operation->database . '.' . $operation->arguments->to,
+                // Note: Database::command() does not inherit WC, so be explicit
+                'writeConcern' => $writeConcern,
             ]];
 
             return $o;
@@ -73,6 +81,7 @@ final class Operation
 
         $o->databaseName = $operation->database;
         $o->collectionName = $operation->collection;
+        $o->collectionOptions = ['writeConcern' => $writeConcern];
         $o->object = self::OBJECT_SELECT_COLLECTION;
 
         return $o;