dr_msg_cb: Added support for message delivery callback

Author: Clement Thuault

conf.c

@@ -45,6 +45,7 @@ void kafka_conf_callbacks_dtor(kafka_conf_callbacks *cbs TSRMLS_DC) /* {{{ */
 {
  kafka_conf_callback_dtor(&cbs->error TSRMLS_CC);
  kafka_conf_callback_dtor(&cbs->rebalance TSRMLS_CC);
+ kafka_conf_callback_dtor(&cbs->dr_msg TSRMLS_CC);
 } /* }}} */
 
 static void kafka_conf_callback_copy(kafka_conf_callback *to, kafka_conf_callback *from TSRMLS_DC) /* {{{ */
@@ -59,6 +60,7 @@ void kafka_conf_callbacks_copy(kafka_conf_callbacks *to, kafka_conf_callbacks *f
 {
  kafka_conf_callback_copy(&to->error, &from->error TSRMLS_CC);
  kafka_conf_callback_copy(&to->rebalance, &from->rebalance TSRMLS_CC);
+ kafka_conf_callback_copy(&to->dr_msg, &from->dr_msg TSRMLS_CC);
 } /* }}} */
 
 static void kafka_conf_free(void *object TSRMLS_DC) /* {{{ */
@@ -158,6 +160,45 @@ static void kafka_conf_error_cb(rd_kafka_t *rk, int err, const char *reason, voi
  zval_ptr_dtor(&zreason);
 }
 
+static void kafka_conf_dr_msg_cb(rd_kafka_t *rk, const rd_kafka_message_t *msg, void *opaque)
+{
+ kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque;
+ zval *retval;
+ zval **args[2];
+ zval *zrk;
+ zval *zrkmsg;
+ TSRMLS_FETCH();
+
+ if (!opaque) {
+ return;
+ }
+
+ if (!cbs->dr_msg.fci.function_name) {
+ return;
+ }
+
+ ALLOC_INIT_ZVAL(zrk);
+ KAFKA_ZVAL_ZVAL(zrk, &cbs->rk, 1, 0);
+
+ ALLOC_INIT_ZVAL(zrkmsg);
+ kafka_message_new(zrkmsg, msg);
+
+ args[0] = &zrk;
+ args[1] = &zrkmsg;
+
+ cbs->dr_msg.fci.retval_ptr_ptr = &retval;
+ cbs->dr_msg.fci.params = args;
+ cbs->dr_msg.fci.param_count = 2;
+
+ zend_call_function(&cbs->dr_msg.fci, &cbs->dr_msg.fcc TSRMLS_CC);
+
+ if (retval) {
+ zval_ptr_dtor(&retval);
+ }
+ zval_ptr_dtor(&zrk);
+ zval_ptr_dtor(&zrkmsg);
+}
+
 #ifdef HAVE_NEW_KAFKA_CONSUMER
 static void kafka_conf_rebalance_cb(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *partitions, void *opaque)
 {
@@ -396,6 +437,41 @@ PHP_METHOD(RdKafka__Conf, setErrorCb)
 }
 /* }}} */
 
+/* {{{ proto void RdKafka\Conf::setDrMsgCb(callable $callback)
+ Sets the delivery report callback */
+
+ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_dr_msg_cb, 0, 0, 1)
+ ZEND_ARG_INFO(0, callback)
+ZEND_END_ARG_INFO()
+
+PHP_METHOD(RdKafka__Conf, setDrMsgCb)
+{
+ zend_fcall_info fci;
+ zend_fcall_info_cache fcc;
+ kafka_conf_object *intern;
+
+ if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "f", &fci, &fcc) == FAILURE) {
+ return;
+ }
+
+ intern = get_kafka_conf_object(getThis() TSRMLS_CC);
+ if (!intern) {
+ return;
+ }
+
+ Z_ADDREF_P(fci.function_name);
+
+ if (intern->cbs.dr_msg.fci.function_name) {
+ zval_ptr_dtor(&intern->cbs.dr_msg.fci.function_name);
+ }
+
+ intern->cbs.dr_msg.fci = fci;
+ intern->cbs.dr_msg.fcc = fcc;
+
+ rd_kafka_conf_set_dr_msg_cb(intern->u.conf, kafka_conf_dr_msg_cb);
+}
+/* }}} */
+
 #ifdef HAVE_NEW_KAFKA_CONSUMER
 /* {{{ proto void RdKafka\Conf::setRebalanceCb(mixed $callback)
  Set rebalance callback for use with coordinated consumer group balancing */
@@ -509,6 +585,7 @@ static const zend_function_entry kafka_conf_fe[] = {
  PHP_ME(RdKafka__Conf, setDefaultTopicConf, arginfo_kafka_conf_set_default_topic_conf, ZEND_ACC_PUBLIC)
 #endif /* HAVE_NEW_KAFKA_CONSUMER */
  PHP_ME(RdKafka__Conf, setErrorCb, arginfo_kafka_conf_set_error_cb, ZEND_ACC_PUBLIC)
+ PHP_ME(RdKafka__Conf, setDrMsgCb, arginfo_kafka_conf_set_dr_msg_cb, ZEND_ACC_PUBLIC)
 #ifdef HAVE_NEW_KAFKA_CONSUMER
  PHP_ME(RdKafka__Conf, setRebalanceCb, arginfo_kafka_conf_set_rebalance_cb, ZEND_ACC_PUBLIC)
 #endif /* HAVE_NEW_KAFKA_CONSUMER */

conf.h

@@ -40,6 +40,7 @@ typedef struct _kafka_conf_callbacks {
  zval rk;
  kafka_conf_callback error;
  kafka_conf_callback rebalance;
+ kafka_conf_callback dr_msg;
 } kafka_conf_callbacks;
 
 typedef struct _kafka_conf_object {

kafka_consumer.c

@@ -658,4 +658,5 @@ void kafka_kafka_consumer_minit(TSRMLS_D) /* {{{ */
 
  zend_declare_property_null(ce, ZEND_STRL("error_cb"), ZEND_ACC_PRIVATE TSRMLS_CC);
  zend_declare_property_null(ce, ZEND_STRL("rebalance_cb"), ZEND_ACC_PRIVATE TSRMLS_CC);
+ zend_declare_property_null(ce, ZEND_STRL("dr_msg_cb"), ZEND_ACC_PRIVATE TSRMLS_CC);
 } /* }}} */

rdkafka.c

@@ -648,6 +648,7 @@ PHP_MINIT_FUNCTION(rdkafka)
  ce_kafka->create_object = kafka_new;
 
  zend_declare_property_null(ce_kafka, ZEND_STRL("error_cb"), ZEND_ACC_PRIVATE TSRMLS_CC);
+ zend_declare_property_null(ce_kafka, ZEND_STRL("dr_cb"), ZEND_ACC_PRIVATE TSRMLS_CC);
 
  INIT_NS_CLASS_ENTRY(ce, "RdKafka", "Consumer", kafka_consumer_fe);
  ce_kafka_consumer = zend_register_internal_class_ex(&ce, ce_kafka, NULL TSRMLS_CC);

tests/conf.phpt

@@ -33,6 +33,11 @@ $conf->setErrorCb(function () { });
 $dump = $conf->dump();
 var_dump(isset($dump["error_cb"]));
 
+echo "Setting dr_msg callback\n";
+$conf->setDrMsgCb(function () { });
+$dump = $conf->dump();
+var_dump(isset($dump["dr_msg_cb"]));
+
 echo "Dumping conf\n";
 var_dump(array_intersect_key($conf->dump(), array(
  "client.id" => true,
@@ -50,6 +55,8 @@ Setting an invalid property
 Caught a RdKafka\Exception: No such configuration property: "invalid"
 Setting error callback
 bool(true)
+Setting dr_msg callback
+bool(true)
 Dumping conf
 array(3) {
  ["client.id"]=>