src/rdkafka_broker.c
rd_kafka_broker_t *
rd_kafka_broker_get_async (rd_kafka_t *rk, int32_t broker_id, int state,
                           rd_kafka_enq_once_t *eonce) {
        int version;
        do {
                rd_kafka_broker_t *rkb;

                version = rd_kafka_brokers_get_state_version(rk);

                rd_kafka_rdlock(rk);
                rkb = rd_kafka_broker_find_by_nodeid0(rk, broker_id, state,
                                                      rd_true);
                rd_kafka_rdunlock(rk);

                if (rkb)
                        return rkb;

        } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce));

        return NULL; /* eonce added to wait list */
}
/**
 * @returns the broker handle fork \p broker_id using cached metadata
 *          information (if available) in state == \p state,
 *          with refcount increaesd.
 *
 *          Otherwise enqueues the \p eonce on the wait-state-change queue
 *          which will be triggered on broker state changes.
 *          It may also be triggered erroneously, so the caller
 *          should call rd_kafka_broker_get_async() again when
 *          the eonce is triggered.
 *
 * @locks none
 * @locality any thread
 */
@returns the broker handle fork \p broker_id using cached metadata
information (if available) in state == \p state,
with refcount increaesd.

Otherwise enqueues the \p eonce on the wait-state-change queue
which will be triggered on broker state changes.
It may also be triggered erroneously, so the caller
should call rd_kafka_broker_get_async() again when
the eonce is triggered.

@locks none
@locality any thread
@returns the broker handle fork \ p broker_id using cached metadata information ( if available ) in state == \ p state with refcount increaesd . Otherwise enqueues the \ p eonce on the wait - state - change queue which will be triggered on broker state changes . It may also be triggered erroneously so the caller should call rd_kafka_broker_get_async () again when the eonce is triggered . @locks none @locality any thread
rd_kafka_broker_t *
rd_kafka_broker_get_async (rd_kafka_t *rk, int32_t broker_id, int state,
                           rd_kafka_enq_once_t *eonce) {
        int version;
        do {
                rd_kafka_broker_t *rkb;
                version = rd_kafka_brokers_get_state_version(rk);
                rd_kafka_rdlock(rk);
                rkb = rd_kafka_broker_find_by_nodeid0(rk, broker_id, state,
                                                      rd_true);
                rd_kafka_rdunlock(rk);
                if (rkb)
                        return rkb;
        } while (!rd_kafka_brokers_wait_state_change_async(rk, version, eonce));
        return NULL; 
}
rd_kafka_broker_t * rd_kafka_broker_get_async ( rd_kafka_t * rk , int32_t broker_id , int state , rd_kafka_enq_once_t * eonce ) { int version ; do { rd_kafka_broker_t * rkb ; version = rd_kafka_brokers_get_state_version ( rk ) ; rd_kafka_rdlock ( rk ) ; rkb = rd_kafka_broker_find_by_nodeid0 ( rk , broker_id , state , rd_true ) ; rd_kafka_rdunlock ( rk ) ; if ( rkb ) return rkb ; } while ( ! rd_kafka_brokers_wait_state_change_async ( rk , version , eonce ) ) ; return NULL ; }
@returns the broker handle fork \p broker_id using cached metadata
          information (if available) in state == \p state,
          with refcount increaesd.
@returns the broker handle fork \ p broker_id using cached metadata information ( if available ) in state == \ p state with refcount increaesd .
/* eonce added to wait list */