PHPC-1804: Manager::addSubscriber and removeSubscriber (#1213)

* Use zend_exception_ce instead of zend_exception_get_default()

zend_exception_get_default() has been deprecated since PHP 7.0. See: php/php-src@f9e9d3a

* Check retval from php_phongo_set_monitoring_callbacks

This ensures that php_phongo_client_register is never called in the event this method fails (however unlikely).

* Index subscriber HashTable by numeric object handles

This also updates add/removeSubscriber to use the new argument parsing macros from deead96.

* PHPC-1804: Manager::addSubscriber and removeSubscriber

Implements per-client event subscribers. Moves existing APM code to a new phongo_apm.c module and adds logic to coordinate dispatching between global and per-client subscribers.

Author: jmikola

config.m4

@@ -105,6 +105,7 @@ if test "$PHP_MONGODB" != "no"; then
     phongo_compat.c \
     src/bson.c \
     src/bson-encode.c \
+    src/phongo_apm.c \
     src/BSON/Binary.c \
     src/BSON/BinaryInterface.c \
     src/BSON/DBPointer.c \

config.w32

@@ -118,7 +118,7 @@ if (PHP_MONGODB != "no") {
   var PHP_MONGODB_MONGOC_SOURCES="mongoc-aggregate.c mongoc-apm.c mongoc-array.c mongoc-async.c mongoc-async-cmd.c mongoc-buffer.c mongoc-bulk-operation.c mongoc-change-stream.c mongoc-client.c mongoc-client-pool.c mongoc-client-session.c mongoc-client-side-encryption.c mongoc-cluster-aws.c mongoc-cluster.c mongoc-cluster-cyrus.c mongoc-cluster-sasl.c mongoc-cluster-sspi.c mongoc-cmd.c mongoc-collection.c mongoc-compression.c mongoc-counters.c mongoc-crypt.c mongoc-crypto.c mongoc-crypto-cng.c mongoc-crypto-common-crypto.c mongoc-crypto-openssl.c mongoc-cursor-array.c mongoc-cursor.c mongoc-cursor-change-stream.c mongoc-cursor-cmd.c mongoc-cursor-cmd-deprecated.c mongoc-cursor-find.c mongoc-cursor-find-cmd.c mongoc-cursor-find-opquery.c mongoc-cursor-legacy.c mongoc-cyrus.c mongoc-database.c mongoc-error.c mongoc-find-and-modify.c mongoc-gridfs-bucket.c mongoc-gridfs-bucket-file.c mongoc-gridfs.c mongoc-gridfs-file.c mongoc-gridfs-file-list.c mongoc-gridfs-file-page.c mongoc-handshake.c mongoc-host-list.c mongoc-http.c mongoc-index.c mongoc-init.c mongoc-interrupt.c mongoc-libressl.c mongoc-linux-distro-scanner.c mongoc-list.c mongoc-log.c mongoc-matcher.c mongoc-matcher-op.c mongoc-memcmp.c mongoc-ocsp-cache.c mongoc-openssl.c mongoc-optional.c mongoc-opts.c mongoc-opts-helpers.c mongoc-queue.c mongoc-rand-cng.c mongoc-rand-common-crypto.c mongoc-rand-openssl.c mongoc-read-concern.c mongoc-read-prefs.c mongoc-rpc.c mongoc-sasl.c mongoc-scram.c mongoc-secure-channel.c mongoc-secure-transport.c mongoc-server-api.c mongoc-server-description.c mongoc-server-monitor.c mongoc-server-stream.c mongoc-set.c mongoc-socket.c mongoc-ssl.c mongoc-sspi.c mongoc-stream-buffered.c mongoc-stream.c mongoc-stream-file.c mongoc-stream-gridfs.c mongoc-stream-gridfs-download.c mongoc-stream-gridfs-upload.c mongoc-stream-socket.c mongoc-stream-tls.c mongoc-stream-tls-libressl.c mongoc-stream-tls-openssl-bio.c mongoc-stream-tls-openssl.c mongoc-stream-tls-secure-channel.c mongoc-stream-tls-secure-transport.c mongoc-timeout.c mongoc-topology-background-monitoring.c mongoc-topology.c mongoc-topology-description-apm.c mongoc-topology-description.c mongoc-topology-scanner.c mongoc-uri.c mongoc-util.c mongoc-version-functions.c mongoc-write-command.c mongoc-write-command-legacy.c mongoc-write-concern.c";
 
   EXTENSION("mongodb", "php_phongo.c phongo_compat.c", null, PHP_MONGODB_CFLAGS);
-  MONGODB_ADD_SOURCES("/src", "bson.c bson-encode.c");
+  MONGODB_ADD_SOURCES("/src", "bson.c bson-encode.c phongo_apm.c");
   MONGODB_ADD_SOURCES("/src/BSON", "Binary.c BinaryInterface.c DBPointer.c Decimal128.c Decimal128Interface.c Int64.c Javascript.c JavascriptInterface.c MaxKey.c MaxKeyInterface.c MinKey.c MinKeyInterface.c ObjectId.c ObjectIdInterface.c Persistable.c Regex.c RegexInterface.c Serializable.c Symbol.c Timestamp.c TimestampInterface.c Type.c Undefined.c Unserializable.c UTCDateTime.c UTCDateTimeInterface.c functions.c");
   MONGODB_ADD_SOURCES("/src/MongoDB", "BulkWrite.c ClientEncryption.c Command.c Cursor.c CursorId.c CursorInterface.c Manager.c Query.c ReadConcern.c ReadPreference.c Server.c ServerApi.c Session.c WriteConcern.c WriteConcernError.c WriteError.c WriteResult.c");
   MONGODB_ADD_SOURCES("/src/MongoDB/Exception", "AuthenticationException.c BulkWriteException.c CommandException.c ConnectionException.c ConnectionTimeoutException.c EncryptionException.c Exception.c ExecutionTimeoutException.c InvalidArgumentException.c LogicException.c RuntimeException.c ServerException.c SSLConnectionException.c UnexpectedValueException.c WriteException.c");

php_phongo.c

@@ -66,6 +66,7 @@
 /* Our stuffz */
 #include "php_phongo.h"
 #include "php_bson.h"
+#include "src/phongo_apm.h"
 #include "src/BSON/functions.h"
 #include "src/MongoDB/Monitoring/functions.h"
 
@@ -2330,181 +2331,6 @@ static bool php_phongo_apply_driver_options_to_uri(mongoc_uri_t* uri, zval* zopt
 }
 #endif
 
-/* APM callbacks */
-static void php_phongo_dispatch_handlers(const char* name, zval* z_event)
-{
-	zval* value;
-
-	ZEND_HASH_FOREACH_VAL_IND(MONGODB_G(subscribers), value)
-	{
-		if (EG(exception)) {
-			break;
-		}
-		/* We can't use the zend_call_method_with_1_params macro here, as it
-		 * does a sizeof() on the name argument, which does only work with
-		 * constant names, but not with parameterized ones as it does
-		 * "sizeof(char*)" in that case. */
-		zend_call_method(PHONGO_COMPAT_OBJ_P(value), NULL, NULL, name, strlen(name), NULL, 1, z_event, NULL);
-	}
-	ZEND_HASH_FOREACH_END();
-}
-
-/* Search for a Manager associated with the given client in the request-scoped
- * registry. If any Manager is found, copy it into the output parameter
- * (incrementing its ref-count) and return true; otherwise, set the output
- * parameter to undefined and return false. */
-static bool php_phongo_copy_manager_for_client(mongoc_client_t* client, zval* out)
-{
-	php_phongo_manager_t* manager;
-
-	if (!MONGODB_G(managers) || zend_hash_num_elements(MONGODB_G(managers)) == 0) {
-		return false;
-	}
-
-	ZEND_HASH_FOREACH_PTR(MONGODB_G(managers), manager)
-	{
-		if (manager->client == client) {
-			ZVAL_OBJ(out, &manager->std);
-			Z_ADDREF_P(out);
-
-			return true;
-		}
-	}
-	ZEND_HASH_FOREACH_END();
-
-	ZVAL_UNDEF(out);
-
-	return false;
-}
-
-static void php_phongo_command_started(const mongoc_apm_command_started_t* event)
-{
-	php_phongo_commandstartedevent_t* p_event;
-	zval                              z_event;
-
-	/* Return early if there are no APM subscribers to notify */
-	if (!MONGODB_G(subscribers) || zend_hash_num_elements(MONGODB_G(subscribers)) == 0) {
-		return;
-	}
-
-	object_init_ex(&z_event, php_phongo_commandstartedevent_ce);
-	p_event = Z_COMMANDSTARTEDEVENT_OBJ_P(&z_event);
-
-	p_event->command_name  = estrdup(mongoc_apm_command_started_get_command_name(event));
-	p_event->server_id     = mongoc_apm_command_started_get_server_id(event);
-	p_event->operation_id  = mongoc_apm_command_started_get_operation_id(event);
-	p_event->request_id    = mongoc_apm_command_started_get_request_id(event);
-	p_event->command       = bson_copy(mongoc_apm_command_started_get_command(event));
-	p_event->database_name = estrdup(mongoc_apm_command_started_get_database_name(event));
-
-	if (!php_phongo_copy_manager_for_client(mongoc_apm_command_started_get_context(event), &p_event->manager)) {
-		phongo_throw_exception(PHONGO_ERROR_UNEXPECTED_VALUE, "Found no Manager for client in APM event context");
-		zval_ptr_dtor(&z_event);
-
-		return;
-	}
-
-	php_phongo_dispatch_handlers("commandStarted", &z_event);
-	zval_ptr_dtor(&z_event);
-}
-
-static void php_phongo_command_succeeded(const mongoc_apm_command_succeeded_t* event)
-{
-	php_phongo_commandsucceededevent_t* p_event;
-	zval                                z_event;
-
-	/* Return early if there are no APM subscribers to notify */
-	if (!MONGODB_G(subscribers) || zend_hash_num_elements(MONGODB_G(subscribers)) == 0) {
-		return;
-	}
-
-	object_init_ex(&z_event, php_phongo_commandsucceededevent_ce);
-	p_event = Z_COMMANDSUCCEEDEDEVENT_OBJ_P(&z_event);
-
-	p_event->command_name    = estrdup(mongoc_apm_command_succeeded_get_command_name(event));
-	p_event->server_id       = mongoc_apm_command_succeeded_get_server_id(event);
-	p_event->operation_id    = mongoc_apm_command_succeeded_get_operation_id(event);
-	p_event->request_id      = mongoc_apm_command_succeeded_get_request_id(event);
-	p_event->duration_micros = mongoc_apm_command_succeeded_get_duration(event);
-	p_event->reply           = bson_copy(mongoc_apm_command_succeeded_get_reply(event));
-
-	if (!php_phongo_copy_manager_for_client(mongoc_apm_command_succeeded_get_context(event), &p_event->manager)) {
-		phongo_throw_exception(PHONGO_ERROR_UNEXPECTED_VALUE, "Found no Manager for client in APM event context");
-		zval_ptr_dtor(&z_event);
-
-		return;
-	}
-
-	php_phongo_dispatch_handlers("commandSucceeded", &z_event);
-	zval_ptr_dtor(&z_event);
-}
-
-static void php_phongo_command_failed(const mongoc_apm_command_failed_t* event)
-{
-	php_phongo_commandfailedevent_t* p_event;
-	zval                             z_event;
-	bson_error_t                     tmp_error = { 0 };
-	zend_class_entry*                default_exception_ce;
-
-	default_exception_ce = zend_exception_get_default();
-
-	/* Return early if there are no APM subscribers to notify */
-	if (!MONGODB_G(subscribers) || zend_hash_num_elements(MONGODB_G(subscribers)) == 0) {
-		return;
-	}
-
-	object_init_ex(&z_event, php_phongo_commandfailedevent_ce);
-	p_event = Z_COMMANDFAILEDEVENT_OBJ_P(&z_event);
-
-	p_event->command_name    = estrdup(mongoc_apm_command_failed_get_command_name(event));
-	p_event->server_id       = mongoc_apm_command_failed_get_server_id(event);
-	p_event->operation_id    = mongoc_apm_command_failed_get_operation_id(event);
-	p_event->request_id      = mongoc_apm_command_failed_get_request_id(event);
-	p_event->duration_micros = mongoc_apm_command_failed_get_duration(event);
-	p_event->reply           = bson_copy(mongoc_apm_command_failed_get_reply(event));
-
-	if (!php_phongo_copy_manager_for_client(mongoc_apm_command_failed_get_context(event), &p_event->manager)) {
-		phongo_throw_exception(PHONGO_ERROR_UNEXPECTED_VALUE, "Found no Manager for client in APM event context");
-		zval_ptr_dtor(&z_event);
-
-		return;
-	}
-
-	/* We need to process and convert the error right here, otherwise
-	 * debug_info will turn into a recursive loop, and with the wrong trace
-	 * locations */
-	mongoc_apm_command_failed_get_error(event, &tmp_error);
-
-	object_init_ex(&p_event->z_error, phongo_exception_from_mongoc_domain(tmp_error.domain, tmp_error.code));
-	zend_update_property_string(default_exception_ce, PHONGO_COMPAT_OBJ_P(&p_event->z_error), ZEND_STRL("message"), tmp_error.message);
-	zend_update_property_long(default_exception_ce, PHONGO_COMPAT_OBJ_P(&p_event->z_error), ZEND_STRL("code"), tmp_error.code);
-
-	php_phongo_dispatch_handlers("commandFailed", &z_event);
-	zval_ptr_dtor(&z_event);
-}
-
-/* Sets the callbacks for APM */
-bool php_phongo_set_monitoring_callbacks(mongoc_client_t* client)
-{
-	bool retval;
-
-	mongoc_apm_callbacks_t* callbacks = mongoc_apm_callbacks_new();
-
-	mongoc_apm_set_command_started_cb(callbacks, php_phongo_command_started);
-	mongoc_apm_set_command_succeeded_cb(callbacks, php_phongo_command_succeeded);
-	mongoc_apm_set_command_failed_cb(callbacks, php_phongo_command_failed);
-
-	retval = mongoc_client_set_apm_callbacks(client, callbacks, client);
-
-	if (!retval) {
-		phongo_throw_exception(PHONGO_ERROR_UNEXPECTED_VALUE, "Failed to set APM callbacks");
-	}
-
-	mongoc_apm_callbacks_destroy(callbacks);
-
-	return retval;
-}
-
 static zval* php_phongo_manager_prepare_manager_for_hash(zval* driverOptions, bool* free)
 {
 	php_phongo_manager_t* manager;
@@ -3485,7 +3311,10 @@ void phongo_manager_init(php_phongo_manager_t* manager, const char* uri_string,
 		goto cleanup;
 	}
 
-	php_phongo_set_monitoring_callbacks(manager->client);
+	if (!phongo_apm_set_callbacks(manager->client)) {
+		/* Exception should already have been thrown */
+		goto cleanup;
+	}
 
 	MONGOC_DEBUG("Created client with hash: %s", manager->client_hash);
 

php_phongo_structs.h

@@ -75,6 +75,7 @@ typedef struct {
 	size_t           client_hash_len;
 	bool             use_persistent_client;
 	zval             key_vault_client_manager;
+	HashTable*       subscribers;
 	zend_object      std;
 } php_phongo_manager_t;
 

src/MongoDB/Manager.c

@@ -27,6 +27,7 @@
 #include "phongo_compat.h"
 #include "php_phongo.h"
 #include "Session.h"
+#include "src/phongo_apm.h"
 
 #define PHONGO_MANAGER_URI_DEFAULT "mongodb://127.0.0.1/"
 
@@ -281,6 +282,28 @@ static PHP_METHOD(Manager, __construct)
 	}
 } /* }}} */
 
+/* {{{ proto void MongoDB\Driver\Manager::addSubscriber(MongoDB\Driver\Monitoring\Subscriber $subscriber)
+   Registers an event subscriber for this Manager */
+static PHP_METHOD(Manager, addSubscriber)
+{
+	php_phongo_manager_t* intern;
+	zval*                 subscriber;
+
+	PHONGO_PARSE_PARAMETERS_START(1, 1)
+	Z_PARAM_OBJECT_OF_CLASS(subscriber, php_phongo_subscriber_ce)
+	PHONGO_PARSE_PARAMETERS_END();
+
+	intern = Z_MANAGER_OBJ_P(getThis());
+
+	/* Lazily initialize the subscriber HashTable */
+	if (!intern->subscribers) {
+		ALLOC_HASHTABLE(intern->subscribers);
+		zend_hash_init(intern->subscribers, 0, NULL, ZVAL_PTR_DTOR, 0);
+	}
+
+	phongo_apm_add_subscriber(intern->subscribers, subscriber);
+} /* }}} */
+
 /* {{{ proto MongoDB\Driver\ClientEncryption MongoDB\Driver\Manager::createClientEncryption(array $options)
    Return a ClientEncryption instance */
 static PHP_METHOD(Manager, createClientEncryption)
@@ -677,6 +700,27 @@ static PHP_METHOD(Manager, getWriteConcern)
 	phongo_writeconcern_init(return_value, mongoc_client_get_write_concern(intern->client));
 } /* }}} */
 
+/* {{{ proto void MongoDB\Driver\Manager::removeSubscriber(MongoDB\Driver\Monitoring\Subscriber $subscriber)
+   Unregisters an event subscriber for this Manager */
+static PHP_METHOD(Manager, removeSubscriber)
+{
+	php_phongo_manager_t* intern;
+	zval*                 subscriber;
+
+	PHONGO_PARSE_PARAMETERS_START(1, 1)
+	Z_PARAM_OBJECT_OF_CLASS(subscriber, php_phongo_subscriber_ce)
+	PHONGO_PARSE_PARAMETERS_END();
+
+	intern = Z_MANAGER_OBJ_P(getThis());
+
+	/* NOP if subscribers HashTable was never initialized by addSubscriber */
+	if (!intern->subscribers) {
+		return;
+	}
+
+	phongo_apm_remove_subscriber(intern->subscribers, subscriber);
+} /* }}} */
+
 /* {{{ proto MongoDB\Driver\Server MongoDB\Driver\Manager::selectServers(MongoDB\Driver\ReadPreference $readPreference)
    Returns a suitable Server for the given ReadPreference */
 static PHP_METHOD(Manager, selectServer)
@@ -787,6 +831,10 @@ ZEND_BEGIN_ARG_INFO_EX(ai_Manager___construct, 0, 0, 0)
 	ZEND_ARG_ARRAY_INFO(0, driverOptions, 0)
 ZEND_END_ARG_INFO()
 
+ZEND_BEGIN_ARG_INFO_EX(ai_Manager_addSubscriber, 0, 0, 1)
+	ZEND_ARG_OBJ_INFO(0, subscriber, MongoDB\\Driver\\Monitoring\\Subscriber, 0)
+ZEND_END_ARG_INFO()
+
 ZEND_BEGIN_ARG_INFO_EX(ai_Manager_createClientEncryption, 0, 0, 1)
 	ZEND_ARG_ARRAY_INFO(0, options, 0)
 ZEND_END_ARG_INFO()
@@ -815,6 +863,10 @@ ZEND_BEGIN_ARG_INFO_EX(ai_Manager_executeBulkWrite, 0, 0, 2)
 	ZEND_ARG_INFO(0, options)
 ZEND_END_ARG_INFO()
 
+ZEND_BEGIN_ARG_INFO_EX(ai_Manager_removeSubscriber, 0, 0, 1)
+	ZEND_ARG_OBJ_INFO(0, subscriber, MongoDB\\Driver\\Monitoring\\Subscriber, 0)
+ZEND_END_ARG_INFO()
+
 ZEND_BEGIN_ARG_INFO_EX(ai_Manager_selectServer, 0, 0, 1)
 	ZEND_ARG_OBJ_INFO(0, readPreference, MongoDB\\Driver\\ReadPreference, 0)
 ZEND_END_ARG_INFO()
@@ -829,6 +881,7 @@ ZEND_END_ARG_INFO()
 static zend_function_entry php_phongo_manager_me[] = {
 	/* clang-format off */
 	PHP_ME(Manager, __construct, ai_Manager___construct, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
+	PHP_ME(Manager, addSubscriber, ai_Manager_addSubscriber, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, createClientEncryption, ai_Manager_createClientEncryption, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, executeCommand, ai_Manager_executeCommand, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, executeReadCommand, ai_Manager_executeRWCommand, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
@@ -840,6 +893,7 @@ static zend_function_entry php_phongo_manager_me[] = {
 	PHP_ME(Manager, getReadPreference, ai_Manager_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, getServers, ai_Manager_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, getWriteConcern, ai_Manager_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
+	PHP_ME(Manager, removeSubscriber, ai_Manager_removeSubscriber, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, selectServer, ai_Manager_selectServer, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Manager, startSession, ai_Manager_startSession, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	ZEND_NAMED_ME(__wakeup, PHP_FN(MongoDB_disabled___wakeup), ai_Manager_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
@@ -880,6 +934,11 @@ static void php_phongo_manager_free_object(zend_object* object) /* {{{ */
 	if (!Z_ISUNDEF(intern->key_vault_client_manager)) {
 		zval_ptr_dtor(&intern->key_vault_client_manager);
 	}
+
+	if (intern->subscribers) {
+		zend_hash_destroy(intern->subscribers);
+		FREE_HASHTABLE(intern->subscribers);
+	}
 } /* }}} */
 
 static zend_object* php_phongo_manager_create_object(zend_class_entry* class_type) /* {{{ */