Merge pull request #678

src/Client.php

@@ -170,9 +170,9 @@ class Client
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
+        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) && ! is_in_transaction($options)) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -246,7 +246,7 @@ class Client
     public function listDatabases(array $options = [])
     {
         $operation = new ListDatabases($options);
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
         return $operation->execute($server);
     }
@@ -307,13 +307,13 @@ class Client
      */
     public function watch(array $pipeline = [], array $options = [])
     {
-        if (! isset($options['readPreference'])) {
+        if (! isset($options['readPreference']) && ! is_in_transaction($options)) {
             $options['readPreference'] = $this->readPreference;
         }
 
-        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern)) {
+        if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern) && ! is_in_transaction($options)) {
             $options['readConcern'] = $this->readConcern;
         }
 

src/Database.php

@@ -198,7 +198,7 @@ class Database
     {
         $hasWriteStage = is_last_pipeline_operator_write($pipeline);
 
-        if (! isset($options['readPreference'])) {
+        if (! isset($options['readPreference']) && ! is_in_transaction($options)) {
             $options['readPreference'] = $this->readPreference;
         }
 
@@ -206,7 +206,7 @@ class Database
             $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
         }
 
-        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
         /* MongoDB 4.2 and later supports a read concern when an $out stage is
          * being used, but earlier versions do not.
@@ -249,7 +249,7 @@ class Database
      */
     public function command($command, array $options = [])
     {
-        if (! isset($options['readPreference'])) {
+        if (! isset($options['readPreference']) && ! is_in_transaction($options)) {
             $options['readPreference'] = $this->readPreference;
         }
 
@@ -258,7 +258,7 @@ class Database
         }
 
         $operation = new DatabaseCommand($this->databaseName, $command, $options);
-        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
         return $operation->execute($server);
     }
@@ -280,9 +280,9 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
+        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) && ! is_in_transaction($options)) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -307,9 +307,9 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
+        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) && ! is_in_transaction($options)) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -335,9 +335,9 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
+        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) && ! is_in_transaction($options)) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -420,7 +420,7 @@ class Database
     public function listCollections(array $options = [])
     {
         $operation = new ListCollections($this->databaseName, $options);
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
         return $operation->execute($server);
     }
@@ -442,9 +442,9 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
+        if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern) && ! is_in_transaction($options)) {
             $options['writeConcern'] = $this->writeConcern;
         }
 
@@ -505,13 +505,13 @@ class Database
      */
     public function watch(array $pipeline = [], array $options = [])
     {
-        if (! isset($options['readPreference'])) {
+        if (! isset($options['readPreference']) && ! is_in_transaction($options)) {
             $options['readPreference'] = $this->readPreference;
         }
 
-        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
+        $server = select_server($this->manager, $options);
 
-        if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern)) {
+        if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern) && ! is_in_transaction($options)) {
             $options['readConcern'] = $this->readConcern;
         }
 

src/Operation/Watch.php

@@ -40,7 +40,6 @@ use function is_object;
 use function is_string;
 use function MongoDB\Driver\Monitoring\addSubscriber;
 use function MongoDB\Driver\Monitoring\removeSubscriber;
-use function MongoDB\extract_session_from_options;
 use function MongoDB\select_server;
 use function MongoDB\server_supports_feature;
 
@@ -381,7 +380,7 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
          * is not usable within transactions, we still check if there is a
          * pinned session. This is to avoid an ambiguous error message about
          * running a command on the wrong server. */
-        $server = select_server($this->manager, $this->aggregateOptions['readPreference'], extract_session_from_options($this->aggregateOptions));
+        $server = select_server($this->manager, $this->aggregateOptions);
 
         $resumeOption = isset($this->changeStreamOptions['startAfter']) && ! $hasAdvanced ? 'startAfter' : 'resumeAfter';
 

src/functions.php

@@ -393,14 +393,40 @@ function extract_session_from_options(array $options)
 }
 
 /**
- * Performs server selection, respecting the server a session may be pinned to
+ * Returns the readPreference option if it is set and valid.
+ *
+ * @internal
+ * @param array $options
+ * @return ReadPreference|null
+ */
+function extract_read_preference_from_options(array $options)
+{
+    if (! isset($options['readPreference']) || ! $options['readPreference'] instanceof ReadPreference) {
+        return null;
+    }
+
+    return $options['readPreference'];
+}
+
+/**
+ * Performs server selection, respecting the readPreference and session options
+ * (if given)
  *
  * @internal
  * @return Server
  */
-function select_server(Manager $manager, ReadPreference $readPreference = null, Session $session = null)
+function select_server(Manager $manager, array $options)
 {
-    $server = $session !== null ? $session->getServer() : null;
+    $session = extract_session_from_options($options);
+    if ($session instanceof Session && $session->getServer() !== null) {
+        return $session->getServer();
+    }
+
+    $readPreference = extract_read_preference_from_options($options);
+    if (! $readPreference instanceof ReadPreference) {
+        // TODO: PHPLIB-476: Read transaction read preference once PHPC-1439 is implemented
+        $readPreference = new ReadPreference(ReadPreference::RP_PRIMARY);
+    }
 
-    return $server ?: $manager->selectServer($readPreference);
+    return $manager->selectServer($readPreference);
 }

tests/SpecTests/TransactionsSpecTest.php

@@ -41,12 +41,6 @@ class TransactionsSpecTest extends FunctionalTestCase
         'transactions/mongos-recovery-token: commitTransaction retry fails on new mongos' => 'isMaster failpoints cannot be disabled',
         'transactions/pin-mongos: remain pinned after non-transient error on commit' => 'Blocked on SPEC-1320',
         'transactions/pin-mongos: unpin after transient error within a transaction and commit' => 'isMaster failpoints cannot be disabled',
-        'transactions/read-pref: default readPreference' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
-        'transactions/read-pref: primary readPreference' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
-        'transactions/run-command: run command with secondary read preference in client option and primary read preference in transaction options' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
-        'transactions/transaction-options: transaction options inherited from client' => 'PHPLIB does not properly inherit readConcern for transactions (PHPLIB-473)',
-        'transactions/transaction-options: readConcern local in defaultTransactionOptions' => 'PHPLIB does not properly inherit readConcern for transactions (PHPLIB-473)',
-        'transactions/transaction-options: readConcern snapshot in startTransaction options' => 'PHPLIB does not properly inherit readConcern for transactions (PHPLIB-473)',
     ];
 
     private function doSetUp()