Skip to content

Commit 9ccfae8

Browse files
jmikolajmikola
committed
PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime
1 parent 0f848cb commit 9ccfae8

File tree

5 files changed

+308
-31
lines changed

5 files changed

+308
-31
lines changed

‎src/Client.php‎

Lines changed: 38 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
use MongoDB\Model\DatabaseInfoIterator;
3030
use MongoDB\Operation\DropDatabase;
3131
use MongoDB\Operation\ListDatabases;
32+
use MongoDB\Operation\Watch;
3233

3334
class Client
3435
{
@@ -37,9 +38,12 @@ class Client
3738
'document' => 'MongoDB\Model\BSONDocument',
3839
'root' => 'MongoDB\Model\BSONDocument',
3940
];
41+
private static $wireVersionForReadConcern = 4;
4042
private static $wireVersionForWritableCommandWriteConcern = 5;
4143

4244
private $manager;
45+
private $readConcern;
46+
private $readPreference;
4347
private $uri;
4448
private $typeMap;
4549
private $writeConcern;
@@ -81,6 +85,8 @@ public function __construct($uri = 'mongodb://127.0.0.1/', array $uriOptions = [
8185
unset($driverOptions['typeMap']);
8286

8387
$this->manager = new Manager($uri, $uriOptions, $driverOptions);
88+
$this->readConcern = $this->manager->getReadConcern();
89+
$this->readPreference = $this->manager->getReadPreference();
8490
$this->writeConcern = $this->manager->getWriteConcern();
8591
}
8692

@@ -173,7 +179,7 @@ public function getManager()
173179
*/
174180
public function getReadConcern()
175181
{
176-
return $this->manager->getReadConcern();
182+
return $this->readConcern;
177183
}
178184

179185
/**
@@ -183,7 +189,7 @@ public function getReadConcern()
183189
*/
184190
public function getReadPreference()
185191
{
186-
return $this->manager->getReadPreference();
192+
return $this->readPreference;
187193
}
188194

189195
/**
@@ -268,4 +274,34 @@ public function startSession(array $options = [])
268274
{
269275
return $this->manager->startSession($options);
270276
}
277+
278+
/**
279+
* Create a change stream for watching changes to the cluster.
280+
*
281+
* @see Watch::__construct() for supported options
282+
* @param array $pipeline List of pipeline operations
283+
* @param array $options Command options
284+
* @return ChangeStream
285+
* @throws InvalidArgumentException for parameter/option parsing errors
286+
*/
287+
public function watch(array $pipeline = [], array $options = [])
288+
{
289+
if ( ! isset($options['readPreference'])) {
290+
$options['readPreference'] = $this->readPreference;
291+
}
292+
293+
$server = $this->manager->selectServer($options['readPreference']);
294+
295+
if ( ! isset($options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
296+
$options['readConcern'] = $this->readConcern;
297+
}
298+
299+
if ( ! isset($options['typeMap'])) {
300+
$options['typeMap'] = $this->typeMap;
301+
}
302+
303+
$operation = new Watch($this->manager, null, null, $pipeline, $options);
304+
305+
return $operation->execute($server);
306+
}
271307
}

‎src/Database.php‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
use MongoDB\Operation\DropDatabase;
3535
use MongoDB\Operation\ListCollections;
3636
use MongoDB\Operation\ModifyCollection;
37+
use MongoDB\Operation\Watch;
3738

3839
class Database
3940
{
@@ -42,6 +43,7 @@ class Database
4243
'document' => 'MongoDB\Model\BSONDocument',
4344
'root' => 'MongoDB\Model\BSONDocument',
4445
];
46+
private static $wireVersionForReadConcern = 4;
4547
private static $wireVersionForWritableCommandWriteConcern = 5;
4648

4749
private $databaseName;
@@ -409,6 +411,36 @@ public function selectGridFSBucket(array $options = [])
409411
return new Bucket($this->manager, $this->databaseName, $options);
410412
}
411413

414+
/**
415+
* Create a change stream for watching changes to the database.
416+
*
417+
* @see Watch::__construct() for supported options
418+
* @param array $pipeline List of pipeline operations
419+
* @param array $options Command options
420+
* @return ChangeStream
421+
* @throws InvalidArgumentException for parameter/option parsing errors
422+
*/
423+
public function watch(array $pipeline = [], array $options = [])
424+
{
425+
if ( ! isset($options['readPreference'])) {
426+
$options['readPreference'] = $this->readPreference;
427+
}
428+
429+
$server = $this->manager->selectServer($options['readPreference']);
430+
431+
if ( ! isset($options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
432+
$options['readConcern'] = $this->readConcern;
433+
}
434+
435+
if ( ! isset($options['typeMap'])) {
436+
$options['typeMap'] = $this->typeMap;
437+
}
438+
439+
$operation = new Watch($this->manager, $this->databaseName, null, $pipeline, $options);
440+
441+
return $operation->execute($server);
442+
}
443+
412444
/**
413445
* Get a clone of this database with different options.
414446
*

‎src/Operation/Watch.php‎

Lines changed: 125 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -18,34 +18,47 @@
1818
namespace MongoDB\Operation;
1919

2020
use MongoDB\ChangeStream;
21+
use MongoDB\BSON\TimestampInterface;
2122
use MongoDB\Driver\Command;
23+
use MongoDB\Driver\Cursor;
2224
use MongoDB\Driver\Manager;
2325
use MongoDB\Driver\ReadConcern;
2426
use MongoDB\Driver\ReadPreference;
2527
use MongoDB\Driver\Server;
2628
use MongoDB\Driver\Session;
2729
use MongoDB\Driver\Exception\RuntimeException;
30+
use MongoDB\Driver\Monitoring\CommandFailedEvent;
31+
use MongoDB\Driver\Monitoring\CommandSubscriber;
32+
use MongoDB\Driver\Monitoring\CommandStartedEvent;
33+
use MongoDB\Driver\Monitoring\CommandSucceededEvent;
2834
use MongoDB\Exception\InvalidArgumentException;
2935
use MongoDB\Exception\UnexpectedValueException;
3036
use MongoDB\Exception\UnsupportedException;
3137

3238
/**
3339
* Operation for creating a change stream with the aggregate command.
3440
*
41+
* Note: the implementation of CommandSubscriber is an internal implementation
42+
* detail and should not be considered part of the public API.
43+
*
3544
* @api
3645
* @see \MongoDB\Collection::watch()
3746
* @see https://docs.mongodb.com/manual/changeStreams/
3847
*/
39-
class Watch implements Executable
48+
class Watch implements Executable, /* @internal */ CommandSubscriber
4049
{
50+
private static $wireVersionForOperationTime = 7;
51+
4152
const FULL_DOCUMENT_DEFAULT = 'default';
4253
const FULL_DOCUMENT_UPDATE_LOOKUP = 'updateLookup';
4354

4455
private $aggregate;
45-
private $databaseName;
56+
private $aggregateOptions;
57+
private $changeStreamOptions;
4658
private $collectionName;
59+
private $databaseName;
60+
private $operationTime;
4761
private $pipeline;
48-
private $options;
4962
private $resumeCallable;
5063

5164
/**
@@ -79,22 +92,44 @@ class Watch implements Executable
7992
* * resumeAfter (document): Specifies the logical starting point for the
8093
* new change stream.
8194
*
95+
* Using this option in conjunction with "startAtOperationTime" will
96+
* result in a server error. The options are mutually exclusive.
97+
*
8298
* * session (MongoDB\Driver\Session): Client session.
8399
*
84100
* Sessions are not supported for server versions < 3.6.
85101
*
102+
* * startAtOperationTime (MongoDB\BSON\TimestampInterface): If specified,
103+
* the change stream will only provide changes that occurred at or after
104+
* the specified timestamp. Any command run against the server will
105+
* return an operation time that can be used here. Alternatively, an
106+
* operation time may be obtained from MongoDB\Driver\Server::getInfo().
107+
*
108+
* Using this option in conjunction with "resumeAfter" will result in a
109+
* server error. The options are mutually exclusive.
110+
*
111+
* This option is not supported for server versions < 4.0.
112+
*
86113
* * typeMap (array): Type map for BSON deserialization. This will be
87114
* applied to the returned Cursor (it is not sent to the server).
88115
*
89-
* @param string $databaseName Database name
90-
* @param string $collectionName Collection name
116+
* Note: A database-level change stream may be created by specifying null
117+
* for the collection name. A cluster-level change stream may be created by
118+
* specifying null for both the database and collection name.
119+
*
120+
* @param Manager $manager Manager instance from the driver
121+
* @param string|null $databaseName Database name
122+
* @param string|null $collectionName Collection name
91123
* @param array $pipeline List of pipeline operations
92124
* @param array $options Command options
93-
* @param Manager $manager Manager instance from the driver
94125
* @throws InvalidArgumentException for parameter/option parsing errors
95126
*/
96127
public function __construct(Manager $manager, $databaseName, $collectionName, array $pipeline, array $options = [])
97128
{
129+
if (isset($collectionName) && ! isset($databaseName)) {
130+
throw new InvalidArgumentException('$collectionName should also be null if $databaseName is null');
131+
}
132+
98133
$options += [
99134
'fullDocument' => self::FULL_DOCUMENT_DEFAULT,
100135
'readPreference' => new ReadPreference(ReadPreference::RP_PRIMARY),
@@ -104,10 +139,12 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
104139
throw InvalidArgumentException::invalidType('"fullDocument" option', $options['fullDocument'], 'string');
105140
}
106141

107-
if (isset($options['resumeAfter'])) {
108-
if ( ! is_array($options['resumeAfter']) && ! is_object($options['resumeAfter'])) {
109-
throw InvalidArgumentException::invalidType('"resumeAfter" option', $options['resumeAfter'], 'array or object');
110-
}
142+
if (isset($options['resumeAfter']) && ! is_array($options['resumeAfter']) && ! is_object($options['resumeAfter'])) {
143+
throw InvalidArgumentException::invalidType('"resumeAfter" option', $options['resumeAfter'], 'array or object');
144+
}
145+
146+
if (isset($options['startAtOperationTime']) && ! $options['startAtOperationTime'] instanceof TimestampInterface) {
147+
throw InvalidArgumentException::invalidType('"startAtOperationTime" option', $options['startAtOperationTime'], TimestampInterface::class);
111148
}
112149

113150
/* In the absence of an explicit session, create one to ensure that the
@@ -122,15 +159,47 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
122159
}
123160
}
124161

162+
$this->aggregateOptions = array_intersect_key($options, ['batchSize' => 1, 'collation' => 1, 'maxAwaitTimeMS' => 1, 'readConcern' => 1, 'readPreference' => 1, 'session' => 1, 'typeMap' => 1]);
163+
$this->changeStreamOptions = array_intersect_key($options, ['fullDocument' => 1, 'resumeAfter' => 1, 'startAtOperationTime' => 1]);
164+
165+
// Null database name implies a cluster-wide change stream
166+
if ($databaseName === null) {
167+
$databaseName = 'admin';
168+
$this->changeStreamOptions['allChangesForCluster'] = true;
169+
}
170+
125171
$this->databaseName = (string) $databaseName;
126-
$this->collectionName = (string) $collectionName;
172+
$this->collectionName = isset($collectionName) ? (string) $collectionName : null;
127173
$this->pipeline = $pipeline;
128-
$this->options = $options;
129174

130175
$this->aggregate = $this->createAggregate();
131176
$this->resumeCallable = $this->createResumeCallable($manager);
132177
}
133178

179+
/** @internal */
180+
final public function commandFailed(CommandFailedEvent $event)
181+
{
182+
}
183+
184+
/** @internal */
185+
final public function commandStarted(CommandStartedEvent $event)
186+
{
187+
}
188+
189+
/** @internal */
190+
final public function commandSucceeded(CommandSucceededEvent $event)
191+
{
192+
if ($event->getCommandName() !== 'aggregate') {
193+
return;
194+
}
195+
196+
$reply = $event->getReply();
197+
198+
if (isset($reply->operationTime) && $reply->operationTime instanceof TimestampInterface) {
199+
$this->operationTime = $reply->operationTime;
200+
}
201+
}
202+
134203
/**
135204
* Execute the operation.
136205
*
@@ -142,47 +211,74 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
142211
*/
143212
public function execute(Server $server)
144213
{
145-
$cursor = $this->aggregate->execute($server);
146-
147-
return new ChangeStream($cursor, $this->resumeCallable);
214+
return new ChangeStream($this->executeAggregate($server), $this->resumeCallable);
148215
}
149216

150217
/**
151218
* Create the aggregate command for creating a change stream.
152219
*
153-
* This method is also used to recreate the aggregate command if a new
154-
* resume token is provided while resuming.
220+
* This method is also used to recreate the aggregate command when resuming.
155221
*
156222
* @return Aggregate
157223
*/
158224
private function createAggregate()
159225
{
160-
$changeStreamOptions = array_intersect_key($this->options, ['fullDocument' => 1, 'resumeAfter' => 1]);
161-
$changeStream = ['$changeStream' => (object) $changeStreamOptions];
162-
163226
$pipeline = $this->pipeline;
164-
array_unshift($pipeline, $changeStream);
227+
array_unshift($pipeline, ['$changeStream' => (object) $this->changeStreamOptions]);
165228

166-
$aggregateOptions = array_intersect_key($this->options, ['batchSize' => 1, 'collation' => 1, 'maxAwaitTimeMS' => 1, 'readConcern' => 1, 'readPreference' => 1, 'session' => 1, 'typeMap' => 1]);
167-
168-
return new Aggregate($this->databaseName, $this->collectionName, $pipeline, $aggregateOptions);
229+
return new Aggregate($this->databaseName, $this->collectionName, $pipeline, $this->aggregateOptions);
169230
}
170231

171232
private function createResumeCallable(Manager $manager)
172233
{
173234
return function($resumeToken = null) use ($manager) {
174-
/* If a resume token was provided, recreate the Aggregate operation
175-
* using the new resume token. */
235+
/* If a resume token was provided, update the "resumeAfter" option
236+
* and ensure that "startAtOperationTime" is no longer set. */
176237
if ($resumeToken !== null) {
177-
$this->options['resumeAfter'] = $resumeToken;
178-
$this->aggregate = $this->createAggregate();
238+
$this->changeStreamOptions['resumeAfter'] = $resumeToken;
239+
unset($this->changeStreamOptions['startAtOperationTime']);
240+
}
241+
242+
/* If we captured an operation time from the first aggregate command
243+
* and there is no "resumeAfter" option, set "startAtOperationTime"
244+
* so that we can resume from the original aggregate's time. */
245+
if ($this->operationTime !== null && ! isset($this->changeStreamOptions['resumeAfter'])) {
246+
$this->changeStreamOptions['startAtOperationTime'] = $this->operationTime;
179247
}
180248

249+
$this->aggregate = $this->createAggregate();
250+
181251
/* Select a new server using the read preference, execute this
182252
* operation on it, and return the new ChangeStream. */
183-
$server = $manager->selectServer($this->options['readPreference']);
253+
$server = $manager->selectServer($this->aggregateOptions['readPreference']);
184254

185255
return $this->execute($server);
186256
};
187257
}
258+
259+
/**
260+
* Execute the aggregate command and optionally capture its operation time.
261+
*
262+
* @param Server $server
263+
* @return Cursor
264+
*/
265+
private function executeAggregate(Server $server)
266+
{
267+
/* If we've already captured an operation time or the server does not
268+
* support returning an operation time (e.g. MongoDB 3.6), execute the
269+
* aggregation directly and return its cursor. */
270+
if ($this->operationTime !== null || ! \MongoDB\server_supports_feature($server, self::$wireVersionForOperationTime)) {
271+
return $this->aggregate->execute($server);
272+
}
273+
274+
/* Otherwise, execute the aggregation using command monitoring so that
275+
* we can capture its operation time with commandSucceeded(). */
276+
\MongoDB\Driver\Monitoring\addSubscriber($this);
277+
278+
try {
279+
return $this->aggregate->execute($server);
280+
} finally {
281+
\MongoDB\Driver\Monitoring\removeSubscriber($this);
282+
}
283+
}
188284
}

0 commit comments

Comments
 (0)