Skip to content

Commit 4c02f94

Browse files
jmikolajmikola
committed
PHPLIB-111: Ensure read ops use appropriate read preference
1 parent a31ca33 commit 4c02f94

File tree

12 files changed

+114
-10
lines changed

12 files changed

+114
-10
lines changed

‎src/Collection.php

Lines changed: 29 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -95,9 +95,16 @@ public function __toString()
9595
*/
9696
public function aggregate(array $pipeline, array $options = array())
9797
{
98-
$readPreference = new ReadPreference(ReadPreference::RP_PRIMARY);
99-
$server = $this->manager->selectServer($readPreference);
98+
if ( ! isset($options['readPreference'])) {
99+
$options['readPreference'] = $this->readPreference;
100+
}
101+
102+
if (\MongoDB\is_last_pipeline_operator_out($pipeline)) {
103+
$options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
104+
}
105+
100106
$operation = new Aggregate($this->databaseName, $this->collectionName, $pipeline, $options);
107+
$server = $this->manager->selectServer($options['readPreference']);
101108

102109
return $operation->execute($server);
103110
}
@@ -132,8 +139,12 @@ public function bulkWrite(array $operations, array $options = array())
132139
*/
133140
public function count($filter = array(), array $options = array())
134141
{
142+
if ( ! isset($options['readPreference'])) {
143+
$options['readPreference'] = $this->readPreference;
144+
}
145+
135146
$operation = new Count($this->databaseName, $this->collectionName, $filter, $options);
136-
$server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
147+
$server = $this->manager->selectServer($options['readPreference']);
137148

138149
return $operation->execute($server);
139150
}
@@ -236,8 +247,12 @@ public function deleteOne($filter, array $options = array())
236247
*/
237248
public function distinct($fieldName, $filter = array(), array $options = array())
238249
{
250+
if ( ! isset($options['readPreference'])) {
251+
$options['readPreference'] = $this->readPreference;
252+
}
253+
239254
$operation = new Distinct($this->databaseName, $this->collectionName, $fieldName, $filter, $options);
240-
$server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
255+
$server = $this->manager->selectServer($options['readPreference']);
241256

242257
return $operation->execute($server);
243258
}
@@ -300,8 +315,12 @@ public function dropIndexes()
300315
*/
301316
public function find($filter = array(), array $options = array())
302317
{
318+
if ( ! isset($options['readPreference'])) {
319+
$options['readPreference'] = $this->readPreference;
320+
}
321+
303322
$operation = new Find($this->databaseName, $this->collectionName, $filter, $options);
304-
$server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
323+
$server = $this->manager->selectServer($options['readPreference']);
305324

306325
return $operation->execute($server);
307326
}
@@ -317,8 +336,12 @@ public function find($filter = array(), array $options = array())
317336
*/
318337
public function findOne($filter = array(), array $options = array())
319338
{
339+
if ( ! isset($options['readPreference'])) {
340+
$options['readPreference'] = $this->readPreference;
341+
}
342+
320343
$operation = new FindOne($this->databaseName, $this->collectionName, $filter, $options);
321-
$server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
344+
$server = $this->manager->selectServer($options['readPreference']);
322345

323346
return $operation->execute($server);
324347
}

‎src/Operation/Aggregate.php

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace MongoDB\Operation;
44

55
use MongoDB\Driver\Command;
6+
use MongoDB\Driver\ReadPreference;
67
use MongoDB\Driver\Server;
78
use MongoDB\Exception\InvalidArgumentException;
89
use MongoDB\Exception\InvalidArgumentTypeException;
@@ -42,6 +43,8 @@ class Aggregate implements Executable
4243
* * maxTimeMS (integer): The maximum amount of time to allow the query to
4344
* run.
4445
*
46+
* * readPreference (MongoDB\Driver\ReadPreference): Read preference.
47+
*
4548
* * useCursor (boolean): Indicates whether the command will request that
4649
* the server provide results using a cursor. The default is true.
4750
*
@@ -94,6 +97,10 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
9497
throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
9598
}
9699

100+
if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
101+
throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
102+
}
103+
97104
if ( ! is_bool($options['useCursor'])) {
98105
throw new InvalidArgumentTypeException('"useCursor" option', $options['useCursor'], 'boolean');
99106
}
@@ -118,9 +125,10 @@ public function __construct($databaseName, $collectionName, array $pipeline, arr
118125
public function execute(Server $server)
119126
{
120127
$isCursorSupported = \MongoDB\server_supports_feature($server, self::$wireVersionForCursor);
128+
$readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
121129

122130
$command = $this->createCommand($server, $isCursorSupported);
123-
$cursor = $server->executeCommand($this->databaseName, $command);
131+
$cursor = $server->executeCommand($this->databaseName, $command, $readPreference);
124132

125133
if ($isCursorSupported && $this->options['useCursor']) {
126134
return $cursor;

‎src/Operation/Count.php

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace MongoDB\Operation;
44

55
use MongoDB\Driver\Command;
6+
use MongoDB\Driver\ReadPreference;
67
use MongoDB\Driver\Server;
78
use MongoDB\Exception\InvalidArgumentException;
89
use MongoDB\Exception\InvalidArgumentTypeException;
@@ -36,6 +37,8 @@ class Count implements Executable
3637
* * maxTimeMS (integer): The maximum amount of time to allow the query to
3738
* run.
3839
*
40+
* * readPreference (MongoDB\Driver\ReadPreference): Read preference.
41+
*
3942
* * skip (integer): The number of documents to skip before returning the
4043
* documents.
4144
*
@@ -69,6 +72,10 @@ public function __construct($databaseName, $collectionName, $filter = array(), a
6972
throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
7073
}
7174

75+
if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
76+
throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
77+
}
78+
7279
if (isset($options['skip']) && ! is_integer($options['skip'])) {
7380
throw new InvalidArgumentTypeException('"skip" option', $options['skip'], 'integer');
7481
}
@@ -88,7 +95,9 @@ public function __construct($databaseName, $collectionName, $filter = array(), a
8895
*/
8996
public function execute(Server $server)
9097
{
91-
$cursor = $server->executeCommand($this->databaseName, $this->createCommand());
98+
$readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
99+
100+
$cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
92101
$result = current($cursor->toArray());
93102

94103
if (empty($result->ok)) {

‎src/Operation/Distinct.php

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace MongoDB\Operation;
44

55
use MongoDB\Driver\Command;
6+
use MongoDB\Driver\ReadPreference;
67
use MongoDB\Driver\Server;
78
use MongoDB\Exception\InvalidArgumentException;
89
use MongoDB\Exception\InvalidArgumentTypeException;
@@ -32,6 +33,8 @@ class Distinct implements Executable
3233
* * maxTimeMS (integer): The maximum amount of time to allow the query to
3334
* run.
3435
*
36+
* * readPreference (MongoDB\Driver\ReadPreference): Read preference.
37+
*
3538
* @param string $databaseName Database name
3639
* @param string $collectionName Collection name
3740
* @param string $fieldName Field for which to return distinct values
@@ -49,6 +52,10 @@ public function __construct($databaseName, $collectionName, $fieldName, $filter
4952
throw new InvalidArgumentTypeException('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
5053
}
5154

55+
if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
56+
throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
57+
}
58+
5259
$this->databaseName = (string) $databaseName;
5360
$this->collectionName = (string) $collectionName;
5461
$this->fieldName = (string) $fieldName;
@@ -65,7 +72,9 @@ public function __construct($databaseName, $collectionName, $fieldName, $filter
6572
*/
6673
public function execute(Server $server)
6774
{
68-
$cursor = $server->executeCommand($this->databaseName, $this->createCommand());
75+
$readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
76+
77+
$cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
6978
$result = current($cursor->toArray());
7079

7180
if (empty($result->ok)) {

‎src/Operation/Find.php

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace MongoDB\Operation;
44

55
use MongoDB\Driver\Query;
6+
use MongoDB\Driver\ReadPreference;
67
use MongoDB\Driver\Server;
78
use MongoDB\Exception\InvalidArgumentException;
89
use MongoDB\Exception\InvalidArgumentTypeException;
@@ -64,6 +65,8 @@ class Find implements Executable
6465
* * projection (document): Limits the fields to return for the matching
6566
* document.
6667
*
68+
* * readPreference (MongoDB\Driver\ReadPreference): Read preference.
69+
*
6770
* * skip (integer): The number of documents to skip before returning.
6871
*
6972
* * sort (document): The order in which to return matching documents. If
@@ -130,6 +133,10 @@ public function __construct($databaseName, $collectionName, $filter, array $opti
130133
throw new InvalidArgumentTypeException('"projection" option', $options['projection'], 'array or object');
131134
}
132135

136+
if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
137+
throw new InvalidArgumentTypeException('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
138+
}
139+
133140
if (isset($options['skip']) && ! is_integer($options['skip'])) {
134141
throw new InvalidArgumentTypeException('"skip" option', $options['skip'], 'integer');
135142
}
@@ -153,7 +160,9 @@ public function __construct($databaseName, $collectionName, $filter, array $opti
153160
*/
154161
public function execute(Server $server)
155162
{
156-
return $server->executeQuery($this->databaseName . '.' . $this->collectionName, $this->createQuery());
163+
$readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
164+
165+
return $server->executeQuery($this->databaseName . '.' . $this->collectionName, $this->createQuery(), $readPreference);
157166
}
158167

159168
/**

‎src/Operation/FindOne.php

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,8 @@ class FindOne implements Executable
3434
* * projection (document): Limits the fields to return for the matching
3535
* document.
3636
*
37+
* * readPreference (MongoDB\Driver\ReadPreference): Read preference.
38+
*
3739
* * skip (integer): The number of documents to skip before returning.
3840
*
3941
* * sort (document): The order in which to return matching documents. If

‎src/functions.php

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,29 @@ function is_first_key_operator($document)
3434
return (isset($firstKey[0]) && $firstKey[0] == '$');
3535
}
3636

37+
/**
38+
* Return whether the aggregation pipeline ends with an $out operator.
39+
*
40+
* This is used for determining whether the aggregation pipeline msut be
41+
* executed against a primary server.
42+
*
43+
* @internal
44+
* @param array $pipeline List of pipeline operations
45+
* @return boolean
46+
*/
47+
function is_last_pipeline_operator_out(array $pipeline)
48+
{
49+
$lastOp = end($pipeline);
50+
51+
if ($lastOp === false) {
52+
return false;
53+
}
54+
55+
$lastOp = (array) $lastOp;
56+
57+
return key($lastOp) === '$out';
58+
}
59+
3760
/**
3861
* Creates a new ReadPreference instance from the Manager's read preference.
3962
*

‎tests/Operation/AggregateTest.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,10 @@ public function provideInvalidConstructorOptions()
4949
$options[][] = array('maxTimeMS' => $value);
5050
}
5151

52+
foreach ($this->getInvalidReadPreferenceValues() as $value) {
53+
$options[][] = array('readPreference' => $value);
54+
}
55+
5256
foreach ($this->getInvalidBooleanValues() as $value) {
5357
$options[][] = array('useCursor' => $value);
5458
}

‎tests/Operation/CountTest.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ public function provideInvalidConstructorOptions()
4040
$options[][] = array('maxTimeMS' => $value);
4141
}
4242

43+
foreach ($this->getInvalidReadPreferenceValues() as $value) {
44+
$options[][] = array('readPreference' => $value);
45+
}
46+
4347
foreach ($this->getInvalidIntegerValues() as $value) {
4448
$options[][] = array('skip' => $value);
4549
}

‎tests/Operation/DistinctTest.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@ public function provideInvalidConstructorOptions()
3232
$options[][] = array('maxTimeMS' => $value);
3333
}
3434

35+
foreach ($this->getInvalidReadPreferenceValues() as $value) {
36+
$options[][] = array('readPreference' => $value);
37+
}
38+
3539
return $options;
3640
}
3741
}

‎tests/Operation/FindTest.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,10 @@ public function provideInvalidConstructorOptions()
6565
$options[][] = array('projection' => $value);
6666
}
6767

68+
foreach ($this->getInvalidReadPreferenceValues() as $value) {
69+
$options[][] = array('readPreference' => $value);
70+
}
71+
6872
foreach ($this->getInvalidIntegerValues() as $value) {
6973
$options[][] = array('skip' => $value);
7074
}

‎tests/Operation/TestCase.php

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,11 @@ protected function getInvalidStringValues()
4040
return array(123, 3.14, true, array(), new stdClass);
4141
}
4242

43+
protected function getInvalidReadPreferenceValues()
44+
{
45+
return array(123, 3.14, 'foo', true, array(), new stdClass);
46+
}
47+
4348
protected function getInvalidWriteConcernValues()
4449
{
4550
return array(123, 3.14, 'foo', true, array(), new stdClass);

0 commit comments

Comments
 (0)