src/rdkafka_broker.c
static int rd_kafka_broker_ops_serve (rd_kafka_broker_t *rkb,
                                      rd_ts_t timeout_us) {
        rd_kafka_op_t *rko;
        int cnt = 0;

        while ((rko = rd_kafka_q_pop(rkb->rkb_ops, timeout_us, 0)) &&
               (cnt++, rd_kafka_broker_op_serve(rkb, rko)))
                timeout_us = RD_POLL_NOWAIT;

        return cnt;
}
/**
 * @brief Serve broker ops.
 * @returns the number of ops served
 */
@brief Serve broker ops.
@returns the number of ops served
@brief Serve broker ops . @returns the number of ops served
static int rd_kafka_broker_ops_serve (rd_kafka_broker_t *rkb,
                                      rd_ts_t timeout_us) {
        rd_kafka_op_t *rko;
        int cnt = 0;
        while ((rko = rd_kafka_q_pop(rkb->rkb_ops, timeout_us, 0)) &&
               (cnt++, rd_kafka_broker_op_serve(rkb, rko)))
                timeout_us = RD_POLL_NOWAIT;
        return cnt;
}
static int rd_kafka_broker_ops_serve ( rd_kafka_broker_t * rkb , rd_ts_t timeout_us ) { rd_kafka_op_t * rko ; int cnt = 0 ; while ( ( rko = rd_kafka_q_pop ( rkb -> rkb_ops , timeout_us , 0 ) ) && ( cnt ++ , rd_kafka_broker_op_serve ( rkb , rko ) ) ) timeout_us = RD_POLL_NOWAIT ; return cnt ; }
@brief Serve broker ops.
@brief Serve broker ops .