conf.c

/*
  +----------------------------------------------------------------------+
  | php-rdkafka                                                          |
  +----------------------------------------------------------------------+
  | Copyright (c) 2016 Arnaud Le Blanc                                   |
  +----------------------------------------------------------------------+
  | This source file is subject to version 3.01 of the PHP license,      |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | http://www.php.net/license/3_01.txt                                  |
  | If you did not receive a copy of the PHP license and are unable to   |
  | obtain it through the world-wide-web, please send a note to          |
  | license@php.net so we can mail you a copy immediately.               |
  +----------------------------------------------------------------------+
  | Author: Arnaud Le Blanc <arnaud.lb@gmail.com>                        |
  +----------------------------------------------------------------------+
*/

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "php.h"
#include "php_ini.h"
#include "ext/standard/info.h"
#include "php_rdkafka.h"
#include "php_rdkafka_priv.h"
#include "librdkafka/rdkafka.h"
#include "Zend/zend_exceptions.h"
#include "ext/spl/spl_exceptions.h"
#include "conf.h"
#include "topic_partition.h"
#include "message.h"
#include "zeval.h"

zend_class_entry * ce_kafka_conf;
zend_class_entry * ce_kafka_topic_conf;

static zend_object_handlers handlers;

static void kafka_conf_callback_dtor(kafka_conf_callback *cb TSRMLS_DC) /* {{{ */
{
    if (cb) {
        zval_ptr_dtor(&cb->fci.function_name);
        efree(cb);
    }
} /* }}} */

void kafka_conf_callbacks_dtor(kafka_conf_callbacks *cbs TSRMLS_DC) /* {{{ */
{
    kafka_conf_callback_dtor(cbs->error TSRMLS_CC);
    cbs->error = NULL;
    kafka_conf_callback_dtor(cbs->rebalance TSRMLS_CC);
    cbs->rebalance = NULL;
    kafka_conf_callback_dtor(cbs->dr_msg TSRMLS_CC);
    cbs->dr_msg = NULL;
    kafka_conf_callback_dtor(cbs->stats TSRMLS_CC);
    cbs->stats = NULL;
    kafka_conf_callback_dtor(cbs->consume TSRMLS_CC);
    cbs->consume = NULL;
    kafka_conf_callback_dtor(cbs->offset_commit TSRMLS_CC);
    cbs->offset_commit = NULL;
    kafka_conf_callback_dtor(cbs->log TSRMLS_CC);
    cbs->log = NULL;
} /* }}} */

static void kafka_conf_callback_copy(kafka_conf_callback **to, kafka_conf_callback *from TSRMLS_DC) /* {{{ */
{
    if (from) {
        *to = emalloc(sizeof(**to));
        **to = *from;
#if PHP_MAJOR_VERSION >= 7
        zval_copy_ctor(&(*to)->fci.function_name);
#else
        Z_ADDREF_P((*to)->fci.function_name);
#endif
    }
} /* }}} */

void kafka_conf_callbacks_copy(kafka_conf_callbacks *to, kafka_conf_callbacks *from TSRMLS_DC) /* {{{ */
{
    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);
    kafka_conf_callback_copy(&to->stats, from->stats TSRMLS_CC);
    kafka_conf_callback_copy(&to->consume, from->consume TSRMLS_CC);
    kafka_conf_callback_copy(&to->offset_commit, from->offset_commit TSRMLS_CC);
    kafka_conf_callback_copy(&to->log, from->log TSRMLS_CC);
} /* }}} */

static void kafka_conf_free(zend_object *object TSRMLS_DC) /* {{{ */
{
    kafka_conf_object *intern = get_custom_object(kafka_conf_object, object);

    switch (intern->type) {
        case KAFKA_CONF:
            if (intern->u.conf) {
                rd_kafka_conf_destroy(intern->u.conf);
            }
            kafka_conf_callbacks_dtor(&intern->cbs TSRMLS_CC);
            break;
        case KAFKA_TOPIC_CONF:
            if (intern->u.topic_conf) {
                rd_kafka_topic_conf_destroy(intern->u.topic_conf);
            }
            break;
    }

    zend_object_std_dtor(&intern->std TSRMLS_CC);

    free_custom_object(intern);
}
/* }}} */

static zend_object_value kafka_conf_new(zend_class_entry *class_type TSRMLS_DC) /* {{{ */
{
    zend_object_value retval;
    kafka_conf_object *intern;

    intern = alloc_object(intern, class_type);
    zend_object_std_init(&intern->std, class_type TSRMLS_CC);
    object_properties_init(&intern->std, class_type);

    STORE_OBJECT(retval, intern, (zend_objects_store_dtor_t) zend_objects_destroy_object, kafka_conf_free, NULL);
    SET_OBJECT_HANDLERS(retval, &handlers);

    return retval;
}
/* }}} */

kafka_conf_object * get_kafka_conf_object(zval *zconf TSRMLS_DC)
{
    kafka_conf_object *oconf = get_custom_object_zval(kafka_conf_object, zconf);

    if (!oconf->type) {
        zend_throw_exception_ex(NULL, 0 TSRMLS_CC, "RdKafka\\Conf::__construct() has not been called" TSRMLS_CC);
        return NULL;
    }

    return oconf;
}

static void kafka_conf_error_cb(rd_kafka_t *rk, int err, const char *reason, void *opaque)
{
    kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque;
    zeval args[3];
    TSRMLS_FETCH();

    if (!opaque) {
        return;
    }

    if (!cbs->error) {
        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);
    MAKE_STD_ZEVAL(args[2]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    ZVAL_LONG(P_ZEVAL(args[1]), err);
    RDKAFKA_ZVAL_STRING(P_ZEVAL(args[2]), reason);

    rdkafka_call_function(&cbs->error->fci, &cbs->error->fcc, NULL, 3, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
    zval_ptr_dtor(&args[2]);
}

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;
    zeval args[2];
    TSRMLS_FETCH();

    if (!opaque) {
        return;
    }

    if (!cbs->dr_msg) {
        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    kafka_message_new(P_ZEVAL(args[1]), msg TSRMLS_CC);

    rdkafka_call_function(&cbs->dr_msg->fci, &cbs->dr_msg->fcc, NULL, 2, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
}

static int kafka_conf_stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque)
{
    kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque;
    zeval args[3];
    TSRMLS_FETCH();

    if (!opaque) {
        return 0;
    }

    if (!cbs->stats) {
        return 0;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);
    MAKE_STD_ZEVAL(args[2]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    RDKAFKA_ZVAL_STRING(P_ZEVAL(args[1]), json);
    ZVAL_LONG(P_ZEVAL(args[2]), json_len);

    rdkafka_call_function(&cbs->stats->fci, &cbs->stats->fcc, NULL, 3, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
    zval_ptr_dtor(&args[2]);

    return 0;
}

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)
{
    kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) opaque;
    zeval args[3];
    TSRMLS_FETCH();

    if (!opaque) {
        return;
    }

    if (!cbs->rebalance) {
        err = rd_kafka_assign(rk, NULL);

        if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
            zend_throw_exception(ce_kafka_exception, rd_kafka_err2str(err), err TSRMLS_CC);
            return;
        }

        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);
    MAKE_STD_ZEVAL(args[2]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    ZVAL_LONG(P_ZEVAL(args[1]), err);
    kafka_topic_partition_list_to_array(P_ZEVAL(args[2]), partitions TSRMLS_CC);

    rdkafka_call_function(&cbs->rebalance->fci, &cbs->rebalance->fcc, NULL, 3, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
    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;
    zeval args[2];
    TSRMLS_FETCH();

    if (!opaque) {
        return;
    }

    if (!cbs->consume) {
        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);

    kafka_message_new(P_ZEVAL(args[0]), msg TSRMLS_CC);
    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[1]), &cbs->zrk, 1, 0);


    rdkafka_call_function(&cbs->consume->fci, &cbs->consume->fcc, NULL, 2, args TSRMLS_CC);

    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;
    zeval args[3];
    TSRMLS_FETCH();

    if (!opaque) {
        return;
    }

    if (!cbs->offset_commit) {
        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);
    MAKE_STD_ZEVAL(args[2]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    ZVAL_LONG(P_ZEVAL(args[1]), err);
    kafka_topic_partition_list_to_array(P_ZEVAL(args[2]), partitions TSRMLS_CC);

    rdkafka_call_function(&cbs->offset_commit->fci, &cbs->offset_commit->fcc, NULL, 3, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
    zval_ptr_dtor(&args[2]);
}

static void kafka_conf_log_cb(const rd_kafka_t *rk, int level, const char *facility, const char *message)
{
    zeval args[4];
    TSRMLS_FETCH();

    kafka_conf_callbacks *cbs = (kafka_conf_callbacks*) rd_kafka_opaque(rk);

    if (!cbs->log) {
        return;
    }

    MAKE_STD_ZEVAL(args[0]);
    MAKE_STD_ZEVAL(args[1]);
    MAKE_STD_ZEVAL(args[2]);
    MAKE_STD_ZEVAL(args[3]);

    KAFKA_ZVAL_ZVAL(P_ZEVAL(args[0]), &cbs->zrk, 1, 0);
    ZVAL_LONG(P_ZEVAL(args[1]), level);
    RDKAFKA_ZVAL_STRING(P_ZEVAL(args[2]), facility);
    RDKAFKA_ZVAL_STRING(P_ZEVAL(args[3]), message);

    rdkafka_call_function(&cbs->log->fci, &cbs->log->fcc, NULL, 4, args TSRMLS_CC);

    zval_ptr_dtor(&args[0]);
    zval_ptr_dtor(&args[1]);
    zval_ptr_dtor(&args[2]);
    zval_ptr_dtor(&args[3]);
}

/* {{{ proto RdKafka\Conf::__construct() */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf___construct, 0, 0, 0)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, __construct)
{
    kafka_conf_object *intern;
    zend_error_handling error_handling;

    zend_replace_error_handling(EH_THROW, spl_ce_InvalidArgumentException, &error_handling TSRMLS_CC);

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "") == FAILURE) {
        zend_restore_error_handling(&error_handling TSRMLS_CC);
        return;
    }

    intern = get_custom_object_zval(kafka_conf_object, getThis());
    intern->type = KAFKA_CONF;
    intern->u.conf = rd_kafka_conf_new();

    zend_restore_error_handling(&error_handling TSRMLS_CC);
}
/* }}} */

/* {{{ proto array RfKafka\Conf::dump()
   Dump the configuration properties and values of `conf` to an array */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_dump, 0, 0, 0)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, dump)
{
    size_t cntp;
    const char **dump;
    kafka_conf_object *intern;
    size_t i;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "") == FAILURE) {
        return;
    }

    intern = get_kafka_conf_object(getThis() TSRMLS_CC);
    if (!intern) {
        return;
    }

    switch (intern->type) {
        case KAFKA_CONF:
            dump = rd_kafka_conf_dump(intern->u.conf, &cntp);
            break;
        case KAFKA_TOPIC_CONF:
            dump = rd_kafka_topic_conf_dump(intern->u.topic_conf, &cntp);
            break;
        default:
            return;
    }

    array_init(return_value);

    for (i = 0; i < cntp; i+=2) {
        const char *key = dump[i];
        const char *value = dump[i+1];
        rdkafka_add_assoc_string(return_value, (char*)key, (char*)value);
    }

    rd_kafka_conf_dump_free(dump, cntp);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::set(string $name, string $value)
   Sets a configuration property. */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set, 0, 0, 2)
    ZEND_ARG_INFO(0, name)
    ZEND_ARG_INFO(0, value)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, set)
{
    char *name;
    arglen_t name_len;
    char *value;
    arglen_t value_len;
    kafka_conf_object *intern;
    rd_kafka_conf_res_t ret = 0;
    char errstr[512];

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "ss", &name, &name_len, &value, &value_len) == FAILURE) {
        return;
    }

    intern = get_kafka_conf_object(getThis() TSRMLS_CC);
    if (!intern) {
        return;
    }

    errstr[0] = '\0';

    switch (intern->type) {
        case KAFKA_CONF:
            ret = rd_kafka_conf_set(intern->u.conf, name, value, errstr, sizeof(errstr));
            break;
        case KAFKA_TOPIC_CONF:
            ret = rd_kafka_topic_conf_set(intern->u.topic_conf, name, value, errstr, sizeof(errstr));
            break;
    }

    switch (ret) {
        case RD_KAFKA_CONF_UNKNOWN:
            zend_throw_exception(ce_kafka_exception, errstr, RD_KAFKA_CONF_UNKNOWN TSRMLS_CC);
            return;
        case RD_KAFKA_CONF_INVALID:
            zend_throw_exception(ce_kafka_exception, errstr, RD_KAFKA_CONF_INVALID TSRMLS_CC);
            return;
        case RD_KAFKA_CONF_OK:
            break;
    }
}
/* }}} */

/* {{{ proto RdKafka\Conf::setDefaultTopicConf(RdKafka\TopicConf $topicConf) */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_default_topic_conf, 0, 0, 1)
    ZEND_ARG_INFO(0, topic_conf)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setDefaultTopicConf)
{
    zval *ztopic_conf;
    kafka_conf_object *intern;
    kafka_conf_object *topic_conf_intern;
    rd_kafka_topic_conf_t *topic_conf;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "O", &ztopic_conf, ce_kafka_topic_conf) == FAILURE) {
        return;
    }

    intern = get_kafka_conf_object(getThis() TSRMLS_CC);
    if (!intern) {
        return;
    }

    topic_conf_intern = get_kafka_conf_object(ztopic_conf TSRMLS_CC);
    if (!topic_conf_intern) {
        return;
    }

    topic_conf = rd_kafka_topic_conf_dup(topic_conf_intern->u.topic_conf);

    rd_kafka_conf_set_default_topic_conf(intern->u.conf, topic_conf);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setErrorCb(callable $callback)
   Sets the error callback */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_error_cb, 0, 0, 1)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setErrorCb)
{
    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(P_ZEVAL(fci.function_name));

    if (intern->cbs.error) {
        zval_ptr_dtor(&intern->cbs.error->fci.function_name);
    } else {
        intern->cbs.error = ecalloc(1, sizeof(*intern->cbs.error));
    }

    intern->cbs.error->fci = fci;
    intern->cbs.error->fcc = fcc;

    rd_kafka_conf_set_error_cb(intern->u.conf, kafka_conf_error_cb);
}
/* }}} */

/* {{{ 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(P_ZEVAL(fci.function_name));

    if (intern->cbs.dr_msg) {
        zval_ptr_dtor(&intern->cbs.dr_msg->fci.function_name);
    } else {
        intern->cbs.dr_msg = ecalloc(1, sizeof(*intern->cbs.dr_msg));
    }

    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);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setStatsCb(callable $callback)
   Sets the statistics report callback */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_stats_cb, 0, 0, 1)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setStatsCb)
{
    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(P_ZEVAL(fci.function_name));

    if (intern->cbs.stats) {
        zval_ptr_dtor(&intern->cbs.stats->fci.function_name);
    } else {
        intern->cbs.stats = ecalloc(1, sizeof(*intern->cbs.stats));
    }

    intern->cbs.stats->fci = fci;
    intern->cbs.stats->fcc = fcc;

    rd_kafka_conf_set_stats_cb(intern->u.conf, kafka_conf_stats_cb);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setRebalanceCb(mixed $callback)
   Set rebalance callback for use with coordinated consumer group balancing */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_rebalance_cb, 0, 0, 1)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setRebalanceCb)
{
    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(P_ZEVAL(fci.function_name));

    if (intern->cbs.rebalance) {
        zval_ptr_dtor(&intern->cbs.rebalance->fci.function_name);
    } else {
        intern->cbs.rebalance = ecalloc(1, sizeof(*intern->cbs.rebalance));
    }

    intern->cbs.rebalance->fci = fci;
    intern->cbs.rebalance->fcc = fcc;

    rd_kafka_conf_set_rebalance_cb(intern->u.conf, kafka_conf_rebalance_cb);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setConsumeCb(callable $callback)
   Set consume callback to use with poll */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_consume_cb, 0, 0, 1)
                ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setConsumeCb)
{
    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(P_ZEVAL(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->u.conf, kafka_conf_consume_cb);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setOffsetCommitCb(mixed $callback)
   Set offset commit callback for use with consumer groups */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_offset_commit_cb, 0, 0, 1)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setOffsetCommitCb)
{
    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(P_ZEVAL(fci.function_name));

    if (intern->cbs.offset_commit) {
        zval_ptr_dtor(&intern->cbs.offset_commit->fci.function_name);
    } else {
        intern->cbs.offset_commit = ecalloc(1, sizeof(*intern->cbs.offset_commit));
    }

    intern->cbs.offset_commit->fci = fci;
    intern->cbs.offset_commit->fcc = fcc;

    rd_kafka_conf_set_offset_commit_cb(intern->u.conf, kafka_conf_offset_commit_cb);
}
/* }}} */

/* {{{ proto void RdKafka\Conf::setLogCb(mixed $callback)
   Set offset commit callback for use with consumer groups */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_conf_set_log_cb, 0, 0, 1)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__Conf, setLogCb)
{
    zend_fcall_info fci;
    zend_fcall_info_cache fcc;
    kafka_conf_object *conf;
    char errstr[512];

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "f", &fci, &fcc) == FAILURE) {
        return;
    }

    conf = get_kafka_conf_object(getThis() TSRMLS_CC);
    if (!conf) {
        return;
    }

    Z_ADDREF_P(P_ZEVAL(fci.function_name));

    if (conf->cbs.log) {
        zval_ptr_dtor(&conf->cbs.log->fci.function_name);
    } else {
        conf->cbs.log = ecalloc(1, sizeof(*conf->cbs.log));
    }

    conf->cbs.log->fci = fci;
    conf->cbs.log->fcc = fcc;

    rd_kafka_conf_set_log_cb(conf->u.conf, kafka_conf_log_cb);
    rd_kafka_conf_set(conf->u.conf, "log.queue", "true", errstr, sizeof(errstr));
}
/* }}} */

/* {{{ proto RdKafka\TopicConf::__construct() */
PHP_METHOD(RdKafka__TopicConf, __construct)
{
    kafka_conf_object *intern;
    zend_error_handling error_handling;

    zend_replace_error_handling(EH_THROW, spl_ce_InvalidArgumentException, &error_handling TSRMLS_CC);

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "") == FAILURE) {
        zend_restore_error_handling(&error_handling TSRMLS_CC);
        return;
    }

    intern = get_custom_object_zval(kafka_conf_object, getThis());
    intern->type = KAFKA_TOPIC_CONF;
    intern->u.topic_conf = rd_kafka_topic_conf_new();

    zend_restore_error_handling(&error_handling TSRMLS_CC);
}
/* }}} */

/* {{{ proto RdKafka\TopicConf::setPartitioner(int $partitioner) */

ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_topic_conf_set_partitioner, 0, 0, 1)
    ZEND_ARG_INFO(0, partitioner)
ZEND_END_ARG_INFO()

PHP_METHOD(RdKafka__TopicConf, setPartitioner)
{
    kafka_conf_object *intern;
    long id;
    int32_t (*partitioner) (const rd_kafka_topic_t * rkt, const void * keydata, size_t keylen, int32_t partition_cnt, void * rkt_opaque, void * msg_opaque);

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l", &id) == FAILURE) {
        return;
    }

    intern = get_kafka_conf_object(getThis() TSRMLS_CC);
    if (!intern) {
        return;
    }

    switch (id) {
        case MSG_PARTITIONER_RANDOM:
            partitioner = rd_kafka_msg_partitioner_random;
            break;
        case MSG_PARTITIONER_CONSISTENT:
            partitioner = rd_kafka_msg_partitioner_consistent;
            break;
        case MSG_PARTITIONER_CONSISTENT_RANDOM:
            partitioner = rd_kafka_msg_partitioner_consistent_random;
            break;
#ifdef HAS_RD_KAFKA_PARTITIONER_MURMUR2
        case MSG_PARTITIONER_MURMUR2:
            partitioner = rd_kafka_msg_partitioner_murmur2;
            break;
        case MSG_PARTITIONER_MURMUR2_RANDOM:
            partitioner = rd_kafka_msg_partitioner_murmur2_random;
            break;
#endif
        default:
            zend_throw_exception_ex(spl_ce_InvalidArgumentException, 0 TSRMLS_CC, "Invalid partitioner given", partitioner TSRMLS_CC);
            return;
    }

    rd_kafka_topic_conf_set_partitioner_cb(intern->u.topic_conf, partitioner);
}
/* }}} */

static const zend_function_entry kafka_topic_conf_fe[] = {
    PHP_ME(RdKafka__TopicConf, __construct, arginfo_kafka_conf___construct, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, dump, arginfo_kafka_conf_dump, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, set, arginfo_kafka_conf_set, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__TopicConf, setPartitioner, arginfo_kafka_topic_conf_set_partitioner, ZEND_ACC_PUBLIC)
    PHP_FE_END
};

static const zend_function_entry kafka_conf_fe[] = {
    PHP_ME(RdKafka__Conf, __construct, arginfo_kafka_conf___construct, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, dump, arginfo_kafka_conf_dump, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, set, arginfo_kafka_conf_set, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, setDefaultTopicConf, arginfo_kafka_conf_set_default_topic_conf, ZEND_ACC_PUBLIC | ZEND_ACC_DEPRECATED)
    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)
    PHP_ME(RdKafka__Conf, setStatsCb, arginfo_kafka_conf_set_stats_cb, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, setRebalanceCb, arginfo_kafka_conf_set_rebalance_cb, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, setConsumeCb, arginfo_kafka_conf_set_consume_cb, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, setOffsetCommitCb, arginfo_kafka_conf_set_offset_commit_cb, ZEND_ACC_PUBLIC)
    PHP_ME(RdKafka__Conf, setLogCb, arginfo_kafka_conf_set_log_cb, ZEND_ACC_PUBLIC)
    PHP_FE_END
};

void kafka_conf_minit(TSRMLS_D)
{
    zend_class_entry tmpce;

    handlers = kafka_default_object_handlers;
    set_object_handler_free_obj(&handlers, kafka_conf_free);
    set_object_handler_offset(&handlers, XtOffsetOf(kafka_conf_object, std));

    INIT_NS_CLASS_ENTRY(tmpce, "RdKafka", "Conf", kafka_conf_fe);
    ce_kafka_conf = zend_register_internal_class(&tmpce TSRMLS_CC);
    ce_kafka_conf->create_object = kafka_conf_new;

    INIT_NS_CLASS_ENTRY(tmpce, "RdKafka", "TopicConf", kafka_topic_conf_fe);
    ce_kafka_topic_conf = zend_register_internal_class(&tmpce TSRMLS_CC);
    ce_kafka_topic_conf->create_object = kafka_conf_new;
}