PHPC-1029 Support maxTimeMS getMore option for tailable command cursors

CONTRIBUTING.md

@@ -37,7 +37,8 @@ URIs and resemble the following:
     "STANDALONE_X509": "mongodb:\/\/C=US,ST=New York,L=New York City,O=MongoDB,OU=KernelUser,CN=client@192.168.112.10:2300\/?authSource=$external&authMechanism=MONGODB-X509",
     "STANDALONE_PLAIN": "mongodb:\/\/root:toor@192.168.112.10:2400\/?authSource=admin",
     "REPLICASET": "mongodb:\/\/192.168.112.10:3000,192.168.112.10:3001,192.168.112.10:3002\/?replicaSet=REPLICASET",
-    "REPLICASET_30": "mongodb:\/\/192.168.112.10:3100,192.168.112.10:3101,192.168.112.10:3102\/?replicaSet=REPLICASET_30"
+    "REPLICASET_30": "mongodb:\/\/192.168.112.10:3100,192.168.112.10:3101,192.168.112.10:3102\/?replicaSet=REPLICASET_30",
+    "REPLICASET_36": "mongodb:\/\/192.168.112.10:3200,192.168.112.10:3201,192.168.112.10:3202\/?replicaSet=REPLICASET_36"
 }
 ```
 

php_phongo.c

@@ -654,7 +654,18 @@ int phongo_execute_command(mongoc_client_t *client, const char *db, zval *zcomma
 	/* According to mongoc_cursor_new_from_command_reply(), the reply bson_t
 	 * is ultimately destroyed on both success and failure. */
 	if (bson_iter_init_find(&iter, &reply, "cursor") && BSON_ITER_HOLDS_DOCUMENT(&iter)) {
-		cmd_cursor = mongoc_cursor_new_from_command_reply(client, &reply, selected_server_id);
+		bson_t initial_reply = BSON_INITIALIZER;
+
+		bson_copy_to(&reply, &initial_reply);
+
+		if (command->max_await_time_ms) {
+			bson_append_bool(&initial_reply, "awaitData", -1, 1);
+			bson_append_int32(&initial_reply, "maxAwaitTimeMS", -1, command->max_await_time_ms);
+			bson_append_bool(&initial_reply, "tailable", -1, 1);
+		}
+
+		cmd_cursor = mongoc_cursor_new_from_command_reply(client, &initial_reply, selected_server_id);
+		bson_destroy(&reply);
 	} else {
 		bson_t *wrapped_reply = create_wrapped_command_envelope(db, &reply);
 
@@ -663,7 +674,7 @@ int phongo_execute_command(mongoc_client_t *client, const char *db, zval *zcomma
 	}
 
 	if (!phongo_advance_cursor_and_check_for_error(cmd_cursor TSRMLS_CC)) {
-		mongoc_cursor_destroy(cmd_cursor);
+		/* If an error is found, the cmd_cursor is destroyed already */
 		return false;
 	}
 

php_phongo_structs.h

@@ -45,7 +45,8 @@ typedef struct {
 
 typedef struct {
 	PHONGO_ZEND_OBJECT_PRE
-	bson_t *bson;
+	bson_t   *bson;
+	uint32_t  max_await_time_ms;
 	PHONGO_ZEND_OBJECT_POST
 } php_phongo_command_t;
 

scripts/presets/replicaset-36.json

@@ -0,0 +1,71 @@
+{
+	"id": "REPLICASET_36",
+	"name": "mongod",
+	"members": [
+		{
+			"procParams": {
+				"dbpath": "/tmp/REPLICASET/3200/",
+				"ipv6": true,
+				"logappend": true,
+				"logpath": "/tmp/REPLICASET/3200/mongod.log",
+				"journal": true,
+				"noprealloc": true,
+				"nssize": 1,
+				"port": 3200,
+				"smallfiles": true,
+				"setParameter": {"enableTestCommands": 1}
+			},
+			"rsParams": {
+				"priority": 99,
+				"tags": {
+					"ordinal": "one",
+					"dc": "pa"
+				}
+			},
+			"server_id": "RS-36-one"
+		},
+		{
+			"procParams": {
+				"dbpath": "/tmp/REPLICASET/3201/",
+				"ipv6": true,
+				"logappend": true,
+				"logpath": "/tmp/REPLICASET/3201/mongod.log",
+				"journal": true,
+				"noprealloc": true,
+				"nssize": 1,
+				"port": 3201,
+				"smallfiles": true,
+				"setParameter": {"enableTestCommands": 1}
+			},
+			"rsParams": {
+				"priority": 1.1,
+				"tags": {
+					"ordinal": "two",
+					"dc": "nyc"
+				}
+			},
+			"server_id": "RS-36-two"
+		},
+		{
+			"procParams": {
+				"dbpath": "/tmp/REPLICASET/3202/",
+				"ipv6": true,
+				"logappend": true,
+				"logpath": "/tmp/REPLICASET/3202/mongod.log",
+				"journal": true,
+				"noprealloc": true,
+				"nssize": 1,
+				"port": 3202,
+				"smallfiles": true,
+				"setParameter": {"enableTestCommands": 1}
+			},
+			"rsParams": {
+				"arbiterOnly": true
+
+			},
+			"server_id": "RS-36-arbiter"
+		}
+	],
+	"version": "36-release"
+}
+

scripts/start-servers.php

@@ -29,6 +29,7 @@ function lap() {
     "replicasets" => [
         "scripts/presets/replicaset.json",
         "scripts/presets/replicaset-30.json",
+        "scripts/presets/replicaset-36.json",
     ],
 ];
 

scripts/ubuntu/mongo-orchestration-config.json

@@ -1,5 +1,6 @@
 {
 	"releases": {
+		"36-release": "/home/vagrant/3.6.0rc3/usr/bin",
 		"32-release": "/home/vagrant/3.2.0/usr/bin",
 		"30-release": "/home/vagrant/3.0.3/usr/bin",
 		"26-release": "/home/vagrant/2.6.9/usr/bin"

scripts/ubuntu/mongo-orchestration.sh

@@ -2,28 +2,32 @@
 apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 7F0CEB10
 # 3.2 key
 apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv EA312927
+# testing key
+apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 58712A2291FA4AD5
 echo 'deb http://repo.mongodb.com/apt/ubuntu precise/mongodb-enterprise/3.2 multiverse' | sudo tee /etc/apt/sources.list.d/mongodb-enterprise-3.2.list
 echo 'deb http://repo.mongodb.com/apt/ubuntu precise/mongodb-enterprise/3.0 multiverse' | sudo tee /etc/apt/sources.list.d/mongodb-enterprise-3.0.list
 echo 'deb http://repo.mongodb.com/apt/ubuntu precise/mongodb-enterprise/2.6 multiverse' | sudo tee /etc/apt/sources.list.d/mongodb-enterprise-2.6.list
+echo 'deb http://repo.mongodb.com/apt/ubuntu precise/mongodb-enterprise/testing multiverse' | sudo tee /etc/apt/sources.list.d/mongodb-enterprise-testing.list
 apt-get update
 
-apt-get install -y libsnmp15 libgsasl7
+apt-get install -y libsnmp15 libgsasl7 libcurl4-openssl-dev
 
 sudo apt-get download mongodb-enterprise-server=3.2.0
 sudo apt-get download mongodb-enterprise-mongos=3.2.0
 sudo apt-get download mongodb-enterprise-server=3.0.3
 sudo apt-get download mongodb-enterprise-server=2.6.9
+sudo apt-get download mongodb-enterprise-server=3.6.0~rc3
 dpkg -x mongodb-enterprise-server_2.6.9_amd64.deb 2.6.9
 dpkg -x mongodb-enterprise-server_3.0.3_amd64.deb 3.0.3
 dpkg -x mongodb-enterprise-server_3.2.0_amd64.deb 3.2.0
 dpkg -x mongodb-enterprise-mongos_3.2.0_amd64.deb 3.2.0
-
+dpkg -x mongodb-enterprise-server_3.6.0~rc3_amd64.deb 3.6.0rc3
 
 
 # Python stuff for mongo-orchestration
 apt-get install -y python python-dev python-pip
 
-pip install --upgrade 'git+https://github.com/10gen/mongo-orchestration.git#egg=mongo_orchestration'
+pip install --index-url=https://pypi.python.org/simple/ --upgrade 'git+https://github.com/10gen/mongo-orchestration.git#egg=mongo_orchestration'
 
 # Launch mongo-orchestration
 mongo-orchestration -f mongo-orchestration-config.json -b 192.168.112.10 --enable-majority-read-concern start

src/MongoDB/Command.c

@@ -21,35 +21,86 @@
 #include <php.h>
 #include <Zend/zend_interfaces.h>
 
+#include "php_array_api.h"
 #include "phongo_compat.h"
 #include "php_phongo.h"
 #include "php_bson.h"
 
 zend_class_entry *php_phongo_command_ce;
 
-/* {{{ proto void MongoDB\Driver\Command::__construct(array|object $document)
+/* Initialize the "maxAwaitTimeMS" option. Returns true on success; otherwise,
+ * false is returned and an exception is thrown.
+ *
+ * The "maxAwaitTimeMS" option is assigned to the cursor after query execution
+ * via mongoc_cursor_set_max_await_time_ms(). */
+static bool php_phongo_command_init_max_await_time_ms(php_phongo_command_t *intern, zval *options TSRMLS_DC) /* {{{ */
+{
+	if (php_array_existsc(options, "maxAwaitTimeMS")) {
+		int64_t max_await_time_ms = php_array_fetchc_long(options, "maxAwaitTimeMS");
+
+		if (max_await_time_ms < 0) {
+			phongo_throw_exception(PHONGO_ERROR_INVALID_ARGUMENT TSRMLS_CC, "Expected \"maxAwaitTimeMS\" option to be >= 0, %" PRId64 " given", max_await_time_ms);
+			return false;
+		}
+
+		if (max_await_time_ms > UINT32_MAX) {
+			phongo_throw_exception(PHONGO_ERROR_INVALID_ARGUMENT TSRMLS_CC, "Expected \"maxAwaitTimeMS\" option to be <= %" PRIu32 ", %" PRId64 " given", UINT32_MAX, max_await_time_ms);
+			return false;
+		}
+
+		intern->max_await_time_ms = (uint32_t) max_await_time_ms;
+	}
+
+	return true;
+} /* }}} */
+
+/* Initializes the php_phongo_command_init from options argument. This
+ * function will fall back to a modifier in the absence of a top-level option
+ * (where applicable). */
+static bool php_phongo_command_init(php_phongo_command_t *intern, zval *filter, zval *options TSRMLS_DC) /* {{{ */
+{
+	intern->bson = bson_new();
+
+	php_phongo_zval_to_bson(filter, PHONGO_BSON_NONE, intern->bson, NULL TSRMLS_CC);
+
+	/* Note: if any exceptions are thrown, we can simply return as PHP will
+	* invoke php_phongo_query_free_object to destruct the object. */
+	if (EG(exception)) {
+		return false;
+	}
+
+	if (!options) {
+		return true;
+	}
+
+	if (!php_phongo_command_init_max_await_time_ms(intern, options TSRMLS_CC)) {
+		return false;
+	}
+
+	return true;
+} /* }}} */
+
+/* {{{ proto void MongoDB\Driver\Command::__construct(array|object $document[, array $options = array()])
    Constructs a new Command */
 static PHP_METHOD(Command, __construct)
 {
 	php_phongo_command_t     *intern;
 	zend_error_handling       error_handling;
 	zval                     *document;
-	bson_t                   *bson = bson_new();
+	zval                     *options = NULL;
 	SUPPRESS_UNUSED_WARNING(return_value) SUPPRESS_UNUSED_WARNING(return_value_ptr) SUPPRESS_UNUSED_WARNING(return_value_used)
 
 
 	zend_replace_error_handling(EH_THROW, phongo_exception_from_phongo_domain(PHONGO_ERROR_INVALID_ARGUMENT), &error_handling TSRMLS_CC);
 	intern = Z_COMMAND_OBJ_P(getThis());
 
-	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "A", &document) == FAILURE) {
+	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "A|a!", &document, &options) == FAILURE) {
 		zend_restore_error_handling(&error_handling TSRMLS_CC);
 		return;
 	}
 	zend_restore_error_handling(&error_handling TSRMLS_CC);
 
-
-	php_phongo_zval_to_bson(document, PHONGO_BSON_NONE, bson, NULL TSRMLS_CC);
-	intern->bson = bson;
+	php_phongo_command_init(intern, document, options TSRMLS_CC);
 } /* }}} */
 
 /* {{{ MongoDB\Driver\Command function entries */

tests/bulk/bulkwrite-update-arrayFilters-001.phpt

@@ -2,12 +2,12 @@
 MongoDB\Driver\BulkWrite::update with arrayFilters
 --SKIPIF--
 <?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
-<?php NEEDS('STANDALONE'); CLEANUP(STANDALONE); ?>
+<?php START('THROWAWAY', ["version" => "36-release"]); CLEANUP(THROWAWAY); ?>
 --FILE--
 <?php
 require_once __DIR__ . "/../utils/basic.inc";
 
-$manager = new MongoDB\Driver\Manager(STANDALONE);
+$manager = new MongoDB\Driver\Manager(THROWAWAY);
 
 $bulk = new MongoDB\Driver\BulkWrite();
 
@@ -33,7 +33,11 @@ $cursor = $manager->executeQuery( DATABASE_NAME . '.' . COLLECTION_NAME, new \Mo
 var_dump($cursor->toArray());
 ?>
 ===DONE===
+<?php DELETE("THROWAWAY"); ?>
 <?php exit(0); ?>
+--CLEAN--
+<?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
+<?php DELETE("THROWAWAY"); ?>
 --EXPECTF--
 array(%d) {
   [0]=>

tests/command/cursor-tailable-001.phpt

@@ -0,0 +1,71 @@
+--TEST--
+MongoDB\Driver\Cursor tailable iteration with awaitData and maxAwaitTimeMS options
+--SKIPIF--
+<?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
+<?php NEEDS('REPLICASET_36'); CLEANUP(REPLICASET_36); ?>
+--FILE--
+<?php
+require_once __DIR__ . "/../utils/basic.inc";
+$manager = new MongoDB\Driver\Manager(REPLICASET_36);
+
+$manager->executeCommand(DATABASE_NAME, new MongoDB\Driver\Command([
+    'create' => COLLECTION_NAME,
+    'capped' => true,
+    'size' => 1048576,
+]));
+
+$bulkWrite = new MongoDB\Driver\BulkWrite;
+$bulkWrite->insert(['_id' => 1]);
+$manager->executeBulkWrite(NS, $bulkWrite);
+
+$pipeline = [
+	[ '$changeStream' => [ 'fullDocument' => 'updateLookup' ] ]
+];
+
+$command = new MongoDB\Driver\Command([
+	'aggregate' => COLLECTION_NAME,
+	'pipeline' => $pipeline,
+	'cursor' => ['batchSize' => 0],
+], [
+    'tailable' => true,
+    'awaitData' => true,
+    'maxAwaitTimeMS' => 100,
+]);
+
+$cursor = $manager->executeCommand(DATABASE_NAME, $command);
+$it = new IteratorIterator($cursor);
+
+$it->rewind();
+$it->next();
+
+$bulkWrite = new MongoDB\Driver\BulkWrite;
+$bulkWrite->insert(['_id' => "new-document"]);
+$manager->executeBulkWrite(NS, $bulkWrite);
+
+$startTime = microtime(true);
+echo "Awaiting results...\n";
+$it->next();
+var_dump($it->current()->operationType, $it->current()->documentKey);
+printf("Waited for %.6f seconds\n", microtime(true) - $startTime);
+
+$startTime = microtime(true);
+echo "Awaiting results...\n";
+$it->next();
+var_dump($it->current()); /* Should be NULL */
+printf("Waited for %.6f seconds\n", microtime(true) - $startTime);
+
+?>
+===DONE===
+<?php exit(0); ?>
+--EXPECTF--
+Awaiting results...
+string(6) "insert"
+object(stdClass)#%d (%d) {
+  ["_id"]=>
+  string(12) "new-document"
+}
+Waited for 0.%d seconds
+Awaiting results...
+NULL
+Waited for 0.1%d seconds
+===DONE===

tests/command/findAndModify-001.phpt

@@ -2,12 +2,12 @@
 MongoDB\Driver\Command with findAndModify and arrayFilters
 --SKIPIF--
 <?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
-<?php NEEDS('STANDALONE'); CLEANUP(STANDALONE); ?>
+<?php START('THROWAWAY', ["version" => "36-release"]); CLEANUP(THROWAWAY); ?>
 --FILE--
 <?php
 require_once __DIR__ . "/../utils/basic.inc";
 
-$manager = new MongoDB\Driver\Manager(STANDALONE);
+$manager = new MongoDB\Driver\Manager(THROWAWAY);
 
 $bulk = new MongoDB\Driver\BulkWrite();
 
@@ -33,7 +33,11 @@ $cursor = $manager->executeQuery( DATABASE_NAME . '.' . COLLECTION_NAME, new \Mo
 var_dump($cursor->toArray());
 ?>
 ===DONE===
+<?php DELETE("THROWAWAY"); ?>
 <?php exit(0); ?>
+--CLEAN--
+<?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
+<?php DELETE("THROWAWAY"); ?>
 --EXPECTF--
 array(%d) {
   [0]=>