plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_admin.c
static rd_kafka_op_t *
rd_kafka_admin_request_op_new (rd_kafka_t *rk,
                               rd_kafka_op_type_t optype,
                               rd_kafka_event_type_t reply_event_type,
                               const struct rd_kafka_admin_worker_cbs *cbs,
                               const rd_kafka_AdminOptions_t *options,
                               rd_kafka_queue_t *rkqu) {
        rd_kafka_op_t *rko;

        rd_assert(rk);
        rd_assert(rkqu);
        rd_assert(cbs);

        rko = rd_kafka_op_new_cb(rk, optype, rd_kafka_admin_worker);

        rko->rko_u.admin_request.reply_event_type = reply_event_type;

        rko->rko_u.admin_request.cbs = (struct rd_kafka_admin_worker_cbs *)cbs;

        /* Make a copy of the options */
        if (options)
                rko->rko_u.admin_request.options = *options;
        else
                rd_kafka_AdminOptions_init(rk,
                                           &rko->rko_u.admin_request.options);

        /* Default to controller */
        rko->rko_u.admin_request.broker_id = -1;

        /* Calculate absolute timeout */
        rko->rko_u.admin_request.abs_timeout =
                rd_timeout_init(
                        rd_kafka_confval_get_int(&rko->rko_u.admin_request.
                                                 options.request_timeout));

        /* Setup enq-op-once, which is triggered by either timer code
         * or future wait-controller code. */
        rko->rko_u.admin_request.eonce =
                rd_kafka_enq_once_new(rko, RD_KAFKA_REPLYQ(rk->rk_ops, 0));

        /* The timer itself must be started from the rdkafka main thread,
         * not here. */

        /* Set up replyq */
        rd_kafka_set_replyq(&rko->rko_u.admin_request.replyq,
                            rkqu->rkqu_q, 0);

        rko->rko_u.admin_request.state = RD_KAFKA_ADMIN_STATE_INIT;
        return rko;
}
/**
 * @brief Create a new admin_request op of type \p optype and sets up the
 *        generic (type independent files).
 *
 *        The caller shall then populate the admin_request.args list
 *        and enqueue the op on rk_ops for further processing work.
 *
 * @param cbs Callbacks, must reside in .data segment.
 * @param options Optional options, may be NULL to use defaults.
 *
 * @locks none
 * @locality application thread
 */
@brief Create a new admin_request op of type \p optype and sets up the
generic (type independent files).

The caller shall then populate the admin_request.args list
and enqueue the op on rk_ops for further processing work.

@param cbs Callbacks, must reside in .data segment.
@param options Optional options, may be NULL to use defaults.

@locks none
@locality application thread
@brief Create a new admin_request op of type \ p optype and sets up the generic ( type independent files ) . The caller shall then populate the admin_request . args list and enqueue the op on rk_ops for further processing work . @param cbs Callbacks must reside in . data segment . @param options Optional options may be NULL to use defaults . @locks none @locality application thread
static rd_kafka_op_t *
rd_kafka_admin_request_op_new (rd_kafka_t *rk,
                               rd_kafka_op_type_t optype,
                               rd_kafka_event_type_t reply_event_type,
                               const struct rd_kafka_admin_worker_cbs *cbs,
                               const rd_kafka_AdminOptions_t *options,
                               rd_kafka_queue_t *rkqu) {
        rd_kafka_op_t *rko;
        rd_assert(rk);
        rd_assert(rkqu);
        rd_assert(cbs);
        rko = rd_kafka_op_new_cb(rk, optype, rd_kafka_admin_worker);
        rko->rko_u.admin_request.reply_event_type = reply_event_type;
        rko->rko_u.admin_request.cbs = (struct rd_kafka_admin_worker_cbs *)cbs;
        if (options)
                rko->rko_u.admin_request.options = *options;
        else
                rd_kafka_AdminOptions_init(rk,
                                           &rko->rko_u.admin_request.options);
        rko->rko_u.admin_request.broker_id = -1;
        rko->rko_u.admin_request.abs_timeout =
                rd_timeout_init(
                        rd_kafka_confval_get_int(&rko->rko_u.admin_request.
                                                 options.request_timeout));
        rko->rko_u.admin_request.eonce =
                rd_kafka_enq_once_new(rko, RD_KAFKA_REPLYQ(rk->rk_ops, 0));
        rd_kafka_set_replyq(&rko->rko_u.admin_request.replyq,
                            rkqu->rkqu_q, 0);
        rko->rko_u.admin_request.state = RD_KAFKA_ADMIN_STATE_INIT;
        return rko;
}
static rd_kafka_op_t * rd_kafka_admin_request_op_new ( rd_kafka_t * rk , rd_kafka_op_type_t optype , rd_kafka_event_type_t reply_event_type , const struct rd_kafka_admin_worker_cbs * cbs , const rd_kafka_AdminOptions_t * options , rd_kafka_queue_t * rkqu ) { rd_kafka_op_t * rko ; rd_assert ( rk ) ; rd_assert ( rkqu ) ; rd_assert ( cbs ) ; rko = rd_kafka_op_new_cb ( rk , optype , rd_kafka_admin_worker ) ; rko -> rko_u . admin_request . reply_event_type = reply_event_type ; rko -> rko_u . admin_request . cbs = ( struct rd_kafka_admin_worker_cbs * ) cbs ; if ( options ) rko -> rko_u . admin_request . options = * options ; else rd_kafka_AdminOptions_init ( rk , & rko -> rko_u . admin_request . options ) ; rko -> rko_u . admin_request . broker_id = -1 ; rko -> rko_u . admin_request . abs_timeout = rd_timeout_init ( rd_kafka_confval_get_int ( & rko -> rko_u . admin_request . options . request_timeout ) ) ; rko -> rko_u . admin_request . eonce = rd_kafka_enq_once_new ( rko , RD_KAFKA_REPLYQ ( rk -> rk_ops , 0 ) ) ; rd_kafka_set_replyq ( & rko -> rko_u . admin_request . replyq , rkqu -> rkqu_q , 0 ) ; rko -> rko_u . admin_request . state = RD_KAFKA_ADMIN_STATE_INIT ; return rko ; }
@brief Create a new admin_request op of type \p optype and sets up the
        generic (type independent files).
@brief Create a new admin_request op of type \ p optype and sets up the generic ( type independent files ) .
/* Make a copy of the options */ /* Default to controller */ /* Calculate absolute timeout */ /* Setup enq-op-once, which is triggered by either timer code
         * or future wait-controller code. */ /* The timer itself must be started from the rdkafka main thread,
         * not here. */ /* Set up replyq */