PHPLIB-316: Migrate operations to use read and write command methods

This also bumps the ext-mongodb dependency to 1.4.0.

composer.json

@@ -13,7 +13,7 @@
         "php": ">=5.5",
         "ext-hash": "*",
         "ext-json": "*",
-        "ext-mongodb": "^1.3.0"
+        "ext-mongodb": "^1.4.0"
     },
     "require-dev": {
         "phpunit/phpunit": "^4.8.36"

src/Operation/Aggregate.php

@@ -233,11 +233,15 @@ class Aggregate implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
+        $hasOutStage = \MongoDB\is_last_pipeline_operator_out($this->pipeline);
         $isCursorSupported = \MongoDB\server_supports_feature($server, self::$wireVersionForCursor);
-        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
 
         $command = $this->createCommand($server, $isCursorSupported);
-        $cursor = $server->executeCommand($this->databaseName, $command, $readPreference);
+        $options = $this->createOptions($hasOutStage);
+
+        $cursor = $hasOutStage
+            ? $server->executeReadWriteCommand($this->databaseName, $command, $options)
+            : $server->executeReadCommand($this->databaseName, $command, $options);
 
         if ($isCursorSupported && $this->options['useCursor']) {
             if (isset($this->options['typeMap'])) {
@@ -299,14 +303,6 @@ class Aggregate implements Executable
             $cmd['hint'] = is_array($this->options['hint']) ? (object) $this->options['hint'] : $this->options['hint'];
         }
 
-        if (isset($this->options['readConcern'])) {
-            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
-        }
-
-        if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
-        }
-
         if ($this->options['useCursor']) {
             $cmd['cursor'] = isset($this->options["batchSize"])
                 ? ['batchSize' => $this->options["batchSize"]]
@@ -315,4 +311,31 @@ class Aggregate implements Executable
 
         return new Command($cmd);
     }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadcommand.php
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadwritecommand.php
+     * @param boolean $hasOutStage
+     * @return array
+     */
+    private function createOptions($hasOutStage)
+    {
+        $options = [];
+
+        if (isset($this->options['readConcern'])) {
+            $options['readConcern'] = $this->options['readConcern'];
+        }
+
+        if ( ! $hasOutStage && isset($this->options['readPreference'])) {
+            $options['readPreference'] = $this->options['readPreference'];
+        }
+
+        if ($hasOutStage && isset($this->options['writeConcern'])) {
+            $options['writeConcern'] = $this->options['writeConcern'];
+        }
+
+        return $options;
+    }
 }

src/Operation/Count.php

@@ -142,9 +142,7 @@ class Count implements Executable
             throw UnsupportedException::readConcernNotSupported();
         }
 
-        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
-
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
+        $cursor = $server->executeReadCommand($this->databaseName, $this->createCommand(), $this->createOptions());
         $result = current($cursor->toArray());
 
         // Older server versions may return a float
@@ -182,10 +180,27 @@ class Count implements Executable
             }
         }
 
+        return new Command($cmd);
+    }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadcommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
         if (isset($this->options['readConcern'])) {
-            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
+            $options['readConcern'] = $this->options['readConcern'];
         }
 
-        return new Command($cmd);
+        if (isset($this->options['readPreference'])) {
+            $options['readPreference'] = $this->options['readPreference'];
+        }
+
+        return $options;
     }
 }

src/Operation/CreateCollection.php

@@ -185,7 +185,7 @@ class CreateCollection implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+        $cursor = $server->executeWriteCommand($this->databaseName, $this->createCommand(), $this->createOptions());
 
         if (isset($this->options['typeMap'])) {
             $cursor->setTypeMap($this->options['typeMap']);
@@ -215,10 +215,23 @@ class CreateCollection implements Executable
             }
         }
 
+        return new Command($cmd);
+    }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executewritecommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
         if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
+            $options['writeConcern'] = $this->options['writeConcern'];
         }
 
-        return new Command($cmd);
+        return $options;
     }
 }

src/Operation/CreateIndexes.php

@@ -143,6 +143,23 @@ class CreateIndexes implements Executable
         return array_map(function(IndexInput $index) { return (string) $index; }, $this->indexes);
     }
 
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executewritecommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
+        if (isset($this->options['writeConcern'])) {
+            $options['writeConcern'] = $this->options['writeConcern'];
+        }
+
+        return $options;
+    }
+
     /**
      * Create one or more indexes for the collection using the createIndexes
      * command.
@@ -161,11 +178,7 @@ class CreateIndexes implements Executable
             $cmd['maxTimeMS'] = $this->options['maxTimeMS'];
         }
 
-        if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
-        }
-
-        $server->executeCommand($this->databaseName, new Command($cmd));
+        $server->executeWriteCommand($this->databaseName, new Command($cmd), $this->createOptions());
     }
 
     /**

src/Operation/Distinct.php

@@ -124,9 +124,7 @@ class Distinct implements Executable
             throw UnsupportedException::readConcernNotSupported();
         }
 
-        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
-
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand(), $readPreference);
+        $cursor = $server->executeReadCommand($this->databaseName, $this->createCommand(), $this->createOptions());
         $result = current($cursor->toArray());
 
         if ( ! isset($result->values) || ! is_array($result->values)) {
@@ -160,10 +158,27 @@ class Distinct implements Executable
             $cmd['maxTimeMS'] = $this->options['maxTimeMS'];
         }
 
+        return new Command($cmd);
+    }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadcommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
         if (isset($this->options['readConcern'])) {
-            $cmd['readConcern'] = \MongoDB\read_concern_as_document($this->options['readConcern']);
+            $options['readConcern'] = $this->options['readConcern'];
         }
 
-        return new Command($cmd);
+        if (isset($this->options['readPreference'])) {
+            $options['readPreference'] = $this->options['readPreference'];
+        }
+
+        return $options;
     }
 }

src/Operation/DropCollection.php

@@ -93,8 +93,10 @@ class DropCollection implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
+        $command = new Command(['drop' => $this->collectionName]);
+
         try {
-            $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+            $cursor = $server->executeWriteCommand($this->databaseName, $command, $this->createOptions());
         } catch (DriverRuntimeException $e) {
             /* The server may return an error if the collection does not exist.
              * Check for an error message (unfortunately, there isn't a code)
@@ -115,18 +117,19 @@ class DropCollection implements Executable
     }
 
     /**
-     * Create the drop command.
+     * Create options for executing the command.
      *
-     * @return Command
+     * @see http://php.net/manual/en/mongodb-driver-server.executewritecommand.php
+     * @return array
      */
-    private function createCommand()
+    private function createOptions()
     {
-        $cmd = ['drop' => $this->collectionName];
+        $options = [];
 
         if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
+            $options['writeConcern'] = $this->options['writeConcern'];
         }
 
-        return new Command($cmd);
+        return $options;
     }
 }

src/Operation/DropDatabase.php

@@ -89,7 +89,8 @@ class DropDatabase implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+        $command = new Command(['dropDatabase' => 1]);
+        $cursor = $server->executeWriteCommand($this->databaseName, $command, $this->createOptions());
 
         if (isset($this->options['typeMap'])) {
             $cursor->setTypeMap($this->options['typeMap']);
@@ -99,18 +100,19 @@ class DropDatabase implements Executable
     }
 
     /**
-     * Create the dropDatabase command.
+     * Create options for executing the command.
      *
-     * @return Command
+     * @see http://php.net/manual/en/mongodb-driver-server.executewritecommand.php
+     * @return array
      */
-    private function createCommand()
+    private function createOptions()
     {
-        $cmd = ['dropDatabase' => 1];
+        $options = [];
 
         if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
+            $options['writeConcern'] = $this->options['writeConcern'];
         }
 
-        return new Command($cmd);
+        return $options;
     }
 }

src/Operation/DropIndexes.php

@@ -107,7 +107,7 @@ class DropIndexes implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand());
+        $cursor = $server->executeWriteCommand($this->databaseName, $this->createCommand(), $this->createOptions());
 
         if (isset($this->options['typeMap'])) {
             $cursor->setTypeMap($this->options['typeMap']);
@@ -132,10 +132,23 @@ class DropIndexes implements Executable
             $cmd['maxTimeMS'] = $this->options['maxTimeMS'];
         }
 
+        return new Command($cmd);
+    }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executewritecommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
         if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
+            $options['writeConcern'] = $this->options['writeConcern'];
         }
 
-        return new Command($cmd);
+        return $options;
     }
 }

src/Operation/FindAndModify.php

@@ -201,7 +201,7 @@ class FindAndModify implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand($server));
+        $cursor = $server->executeReadWriteCommand($this->databaseName, $this->createCommand($server), $this->createOptions());
         $result = current($cursor->toArray());
 
         if ( ! isset($result->value)) {
@@ -265,10 +265,23 @@ class FindAndModify implements Executable
             $cmd['bypassDocumentValidation'] = $this->options['bypassDocumentValidation'];
         }
 
+        return new Command($cmd);
+    }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadwritecommand.php
+     * @return array
+     */
+    private function createOptions()
+    {
+        $options = [];
+
         if (isset($this->options['writeConcern'])) {
-            $cmd['writeConcern'] = \MongoDB\write_concern_as_document($this->options['writeConcern']);
+            $options['writeConcern'] = $this->options['writeConcern'];
         }
 
-        return new Command($cmd);
+        return $options;
     }
 }

src/Operation/MapReduce.php

@@ -243,8 +243,15 @@ class MapReduce implements Executable
             throw UnsupportedException::writeConcernNotSupported();
         }
 
-        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
-        $cursor = $server->executeCommand($this->databaseName, $this->createCommand($server), $readPreference);
+        $hasOutputCollection = ! \MongoDB\is_mapreduce_output_inline($this->out);
+
+        $command = $this->createCommand($server);
+        $options = $this->createOptions($hasOutputCollection);
+
+        $cursor = $hasOutputCollection
+            ? $server->executeReadWriteCommand($this->databaseName, $command, $options)
+            : $server->executeReadCommand($this->databaseName, $command, $options);
+
         $result = current($cursor->toArray());
 
         $getIterator = $this->createGetIteratorCallable($result, $server);
@@ -267,7 +274,7 @@ class MapReduce implements Executable
             'out' => $this->out,
         ];
 
-        foreach (['finalize', 'jsMode', 'limit', 'maxTimeMS', 'readConcern', 'verbose', 'writeConcern'] as $option) {
+        foreach (['finalize', 'jsMode', 'limit', 'maxTimeMS', 'verbose'] as $option) {
             if (isset($this->options[$option])) {
                 $cmd[$option] = $this->options[$option];
             }
@@ -323,4 +330,31 @@ class MapReduce implements Executable
 
         throw new UnexpectedValueException('mapReduce command did not return inline results or an output collection');
     }
+
+    /**
+     * Create options for executing the command.
+     *
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadcommand.php
+     * @see http://php.net/manual/en/mongodb-driver-server.executereadwritecommand.php
+     * @param boolean $hasOutputCollection
+     * @return array
+     */
+    private function createOptions($hasOutputCollection)
+    {
+        $options = [];
+
+        if (isset($this->options['readConcern'])) {
+            $options['readConcern'] = $this->options['readConcern'];
+        }
+
+        if ( ! $hasOutputCollection && isset($this->options['readPreference'])) {
+            $options['readPreference'] = $this->options['readPreference'];
+        }
+
+        if ($hasOutputCollection && isset($this->options['writeConcern'])) {
+            $options['writeConcern'] = $this->options['writeConcern'];
+        }
+
+        return $options;
+    }
 }