From a2394c59594583ad1c25d883dfdadf0d38704929 Mon Sep 17 00:00:00 2001 From: nick Date: Mon, 28 Dec 2020 02:50:24 +0100 Subject: [PATCH 1/2] remove legacy consume cb --- configuration.c | 61 ----------------------------------------- configuration.stub.php | 2 -- configuration_arginfo.h | 6 +--- 3 files changed, 1 insertion(+), 68 deletions(-) diff --git a/configuration.c b/configuration.c index d784077..f8e88cb 100644 --- a/configuration.c +++ b/configuration.c @@ -31,8 +31,6 @@ void kafka_conf_callbacks_dtor(kafka_conf_callbacks *cbs) /* {{{ */ cbs->dr_msg = NULL; kafka_conf_callback_dtor(cbs->stats); cbs->stats = NULL; - kafka_conf_callback_dtor(cbs->consume); - cbs->consume = NULL; kafka_conf_callback_dtor(cbs->offset_commit); cbs->offset_commit = NULL; kafka_conf_callback_dtor(cbs->log); @@ -54,7 +52,6 @@ void kafka_conf_callbacks_copy(kafka_conf_callbacks *to, kafka_conf_callbacks *f kafka_conf_callback_copy(&to->rebalance, from->rebalance); kafka_conf_callback_copy(&to->dr_msg, from->dr_msg); kafka_conf_callback_copy(&to->stats, from->stats); - kafka_conf_callback_copy(&to->consume, from->consume); kafka_conf_callback_copy(&to->offset_commit, from->offset_commit); kafka_conf_callback_copy(&to->log, from->log); } /* }}} */ @@ -218,32 +215,6 @@ static void kafka_conf_rebalance_cb(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_ zval_ptr_dtor(&args[2]); } -static void kafka_conf_consume_cb(rd_kafka_message_t *msg, void *opaque) -{ - kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque; - zval args[2]; - - if (!opaque) { - return; - } - - if (!cbs->consume) { - return; - } - - ZVAL_NULL(&args[0]); - ZVAL_NULL(&args[1]); - - kafka_message_new(&args[0], msg); - ZVAL_ZVAL(&args[1], &cbs->zrk, 1, 0); - - - kafka_call_function(&cbs->consume->fci, &cbs->consume->fcc, NULL, 2, args); - - zval_ptr_dtor(&args[0]); - zval_ptr_dtor(&args[1]); -} - static void kafka_conf_offset_commit_cb(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *partitions, void *opaque) { kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque; @@ -509,38 +480,6 @@ ZEND_METHOD(Kafka_Configuration, setRebalanceCb) } /* }}} */ -/* {{{ proto void Kafka\Configuration::setConsumeCb(callable $callback) - Set consume callback to use with poll */ -ZEND_METHOD(Kafka_Configuration, setConsumeCb) -{ - zend_fcall_info fci; - zend_fcall_info_cache fcc; - kafka_conf_object *intern; - - ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 1, 1) - Z_PARAM_FUNC(fci, fcc) - ZEND_PARSE_PARAMETERS_END(); - - intern = get_kafka_conf_object(getThis()); - if (!intern) { - return; - } - - Z_ADDREF_P(&fci.function_name); - - if (intern->cbs.consume) { - zval_ptr_dtor(&intern->cbs.consume->fci.function_name); - } else { - intern->cbs.consume = ecalloc(1, sizeof(*intern->cbs.consume)); - } - - intern->cbs.consume->fci = fci; - intern->cbs.consume->fcc = fcc; - - rd_kafka_conf_set_consume_cb(intern->conf, kafka_conf_consume_cb); -} -/* }}} */ - /* {{{ proto void Kafka\Configuration::setOffsetCommitCb(callback $callback) Set offset commit callback for use with consumer groups */ ZEND_METHOD(Kafka_Configuration, setOffsetCommitCb) diff --git a/configuration.stub.php b/configuration.stub.php index a4c0b2f..e19517c 100644 --- a/configuration.stub.php +++ b/configuration.stub.php @@ -20,8 +20,6 @@ public function setStatsCb(callable $callback): void {} public function setRebalanceCb(callable $callback): void {} - public function setConsumeCb(callable $callback): void {} - public function setOffsetCommitCb(callable $callback): void {} public function setLogCb(callable $callback): void {} diff --git a/configuration_arginfo.h b/configuration_arginfo.h index eddae14..1d7941a 100644 --- a/configuration_arginfo.h +++ b/configuration_arginfo.h @@ -1,5 +1,5 @@ /* This is a generated file, edit the .stub.php file instead. - * Stub hash: 8077a6ec354f5333de7ed2e808e970a93f1a5c89 */ + * Stub hash: 2a9fd8de9f13ef85f4c02373cc6f2dc87114edd0 */ ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Kafka_Configuration___construct, 0, 0, 0) ZEND_END_ARG_INFO() @@ -22,8 +22,6 @@ ZEND_END_ARG_INFO() #define arginfo_class_Kafka_Configuration_setRebalanceCb arginfo_class_Kafka_Configuration_setErrorCb -#define arginfo_class_Kafka_Configuration_setConsumeCb arginfo_class_Kafka_Configuration_setErrorCb - #define arginfo_class_Kafka_Configuration_setOffsetCommitCb arginfo_class_Kafka_Configuration_setErrorCb #define arginfo_class_Kafka_Configuration_setLogCb arginfo_class_Kafka_Configuration_setErrorCb @@ -36,7 +34,6 @@ ZEND_METHOD(Kafka_Configuration, setErrorCb); ZEND_METHOD(Kafka_Configuration, setDrMsgCb); ZEND_METHOD(Kafka_Configuration, setStatsCb); ZEND_METHOD(Kafka_Configuration, setRebalanceCb); -ZEND_METHOD(Kafka_Configuration, setConsumeCb); ZEND_METHOD(Kafka_Configuration, setOffsetCommitCb); ZEND_METHOD(Kafka_Configuration, setLogCb); @@ -49,7 +46,6 @@ static const zend_function_entry class_Kafka_Configuration_methods[] = { ZEND_ME(Kafka_Configuration, setDrMsgCb, arginfo_class_Kafka_Configuration_setDrMsgCb, ZEND_ACC_PUBLIC) ZEND_ME(Kafka_Configuration, setStatsCb, arginfo_class_Kafka_Configuration_setStatsCb, ZEND_ACC_PUBLIC) ZEND_ME(Kafka_Configuration, setRebalanceCb, arginfo_class_Kafka_Configuration_setRebalanceCb, ZEND_ACC_PUBLIC) - ZEND_ME(Kafka_Configuration, setConsumeCb, arginfo_class_Kafka_Configuration_setConsumeCb, ZEND_ACC_PUBLIC) ZEND_ME(Kafka_Configuration, setOffsetCommitCb, arginfo_class_Kafka_Configuration_setOffsetCommitCb, ZEND_ACC_PUBLIC) ZEND_ME(Kafka_Configuration, setLogCb, arginfo_class_Kafka_Configuration_setLogCb, ZEND_ACC_PUBLIC) ZEND_FE_END From 286561f10e7b478d1dcbf378552c1f415f6f2015 Mon Sep 17 00:00:00 2001 From: nick Date: Mon, 28 Dec 2020 02:56:02 +0100 Subject: [PATCH 2/2] fix tests --- tests/conf_callbacks.phpt | 7 ------- 1 file changed, 7 deletions(-) diff --git a/tests/conf_callbacks.phpt b/tests/conf_callbacks.phpt index 3c74567..a3e0156 100644 --- a/tests/conf_callbacks.phpt +++ b/tests/conf_callbacks.phpt @@ -6,11 +6,6 @@ Kafka\Configuration $conf = new Kafka\Configuration(); -echo "Setting consume callback\n"; -$conf->setConsumeCb(function () { }); -$dump = $conf->dump(); -var_dump(isset($dump["consume_cb"])); - echo "Setting offset_commit callback\n"; $conf->setOffsetCommitCb(function () { }); $dump = $conf->dump(); @@ -23,8 +18,6 @@ var_dump(isset($dump["rebalance_cb"])); --EXPECT-- -Setting consume callback -bool(true) Setting offset_commit callback bool(true) Setting rebalance callback