src/rdkafka.c
static rd_kafka_message_t *rd_kafka_consume0 (rd_kafka_t *rk,
                                              rd_kafka_q_t *rkq,
					      int timeout_ms) {
	rd_kafka_op_t *rko;
	rd_kafka_message_t *rkmessage = NULL;
	rd_ts_t abs_timeout = rd_timeout_init(timeout_ms);

        if (timeout_ms)
                rd_kafka_app_poll_blocking(rk);

	rd_kafka_yield_thread = 0;
        while ((rko = rd_kafka_q_pop(rkq,
                                     rd_timeout_remains_us(abs_timeout), 0))) {
                rd_kafka_op_res_t res;

                res = rd_kafka_poll_cb(rk, rkq, rko,
                                       RD_KAFKA_Q_CB_RETURN, NULL);

                if (res == RD_KAFKA_OP_RES_PASS)
                        break;

                if (unlikely(res == RD_KAFKA_OP_RES_YIELD ||
                             rd_kafka_yield_thread)) {
                        /* Callback called rd_kafka_yield(), we must
                         * stop dispatching the queue and return. */
                        rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__INTR,
                                                EINTR);
                        rd_kafka_app_polled(rk);
                        return NULL;
                }

                /* Message was handled by callback. */
                continue;
        }

	if (!rko) {
		/* Timeout reached with no op returned. */
		rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__TIMED_OUT,
					ETIMEDOUT);
                rd_kafka_app_polled(rk);
		return NULL;
	}

        rd_kafka_assert(rk,
                        rko->rko_type == RD_KAFKA_OP_FETCH ||
                        rko->rko_type == RD_KAFKA_OP_CONSUMER_ERR);

	/* Get rkmessage from rko */
	rkmessage = rd_kafka_message_get(rko);

	/* Store offset */
	rd_kafka_op_offset_store(rk, rko, rkmessage);

	rd_kafka_set_last_error(0, 0);

        rd_kafka_app_polled(rk);

	return rkmessage;
}
/**
 * Serve queue 'rkq' and return one message.
 * By serving the queue it will also call any registered callbacks
 * registered for matching events, this includes consumer_cb()
 * in which case no message will be returned.
 */
Serve queue 'rkq' and return one message.
By serving the queue it will also call any registered callbacks
registered for matching events, this includes consumer_cb()
in which case no message will be returned.
Serve queue ' rkq ' and return one message . By serving the queue it will also call any registered callbacks registered for matching events this includes consumer_cb () in which case no message will be returned .
static rd_kafka_message_t *rd_kafka_consume0 (rd_kafka_t *rk,
                                              rd_kafka_q_t *rkq,
					      int timeout_ms) {
	rd_kafka_op_t *rko;
	rd_kafka_message_t *rkmessage = NULL;
	rd_ts_t abs_timeout = rd_timeout_init(timeout_ms);
        if (timeout_ms)
                rd_kafka_app_poll_blocking(rk);
	rd_kafka_yield_thread = 0;
        while ((rko = rd_kafka_q_pop(rkq,
                                     rd_timeout_remains_us(abs_timeout), 0))) {
                rd_kafka_op_res_t res;
                res = rd_kafka_poll_cb(rk, rkq, rko,
                                       RD_KAFKA_Q_CB_RETURN, NULL);
                if (res == RD_KAFKA_OP_RES_PASS)
                        break;
                if (unlikely(res == RD_KAFKA_OP_RES_YIELD ||
                             rd_kafka_yield_thread)) {
                        rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__INTR,
                                                EINTR);
                        rd_kafka_app_polled(rk);
                        return NULL;
                }
                continue;
        }
	if (!rko) {
		rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__TIMED_OUT,
					ETIMEDOUT);
                rd_kafka_app_polled(rk);
		return NULL;
	}
        rd_kafka_assert(rk,
                        rko->rko_type == RD_KAFKA_OP_FETCH ||
                        rko->rko_type == RD_KAFKA_OP_CONSUMER_ERR);
	rkmessage = rd_kafka_message_get(rko);
	rd_kafka_op_offset_store(rk, rko, rkmessage);
	rd_kafka_set_last_error(0, 0);
        rd_kafka_app_polled(rk);
	return rkmessage;
}
static rd_kafka_message_t * rd_kafka_consume0 ( rd_kafka_t * rk , rd_kafka_q_t * rkq , int timeout_ms ) { rd_kafka_op_t * rko ; rd_kafka_message_t * rkmessage = NULL ; rd_ts_t abs_timeout = rd_timeout_init ( timeout_ms ) ; if ( timeout_ms ) rd_kafka_app_poll_blocking ( rk ) ; rd_kafka_yield_thread = 0 ; while ( ( rko = rd_kafka_q_pop ( rkq , rd_timeout_remains_us ( abs_timeout ) , 0 ) ) ) { rd_kafka_op_res_t res ; res = rd_kafka_poll_cb ( rk , rkq , rko , RD_KAFKA_Q_CB_RETURN , NULL ) ; if ( res == RD_KAFKA_OP_RES_PASS ) break ; if ( unlikely ( res == RD_KAFKA_OP_RES_YIELD || rd_kafka_yield_thread ) ) { rd_kafka_set_last_error ( RD_KAFKA_RESP_ERR__INTR , EINTR ) ; rd_kafka_app_polled ( rk ) ; return NULL ; } continue ; } if ( ! rko ) { rd_kafka_set_last_error ( RD_KAFKA_RESP_ERR__TIMED_OUT , ETIMEDOUT ) ; rd_kafka_app_polled ( rk ) ; return NULL ; } rd_kafka_assert ( rk , rko -> rko_type == RD_KAFKA_OP_FETCH || rko -> rko_type == RD_KAFKA_OP_CONSUMER_ERR ) ; rkmessage = rd_kafka_message_get ( rko ) ; rd_kafka_op_offset_store ( rk , rko , rkmessage ) ; rd_kafka_set_last_error ( 0 , 0 ) ; rd_kafka_app_polled ( rk ) ; return rkmessage ; }
Serve queue 'rkq' and return one message.
Serve queue ' rkq ' and return one message .
/* Callback called rd_kafka_yield(), we must
                         * stop dispatching the queue and return. */ /* Message was handled by callback. */ /* Timeout reached with no op returned. */ /* Get rkmessage from rko */ /* Store offset */