def _basic_consume_impl(self,
                            queue,
                            no_ack,
                            exclusive,
                            consumer_tag,
                            arguments=None,
                            consumer_callback=None,
                            alternate_event_sink=None):
        if (consumer_callback is None) == (alternate_event_sink is None):
            raise ValueError(
                ('exactly one of consumer_callback/alternate_event_sink must '
                 'be non-None', consumer_callback, alternate_event_sink))

        if not consumer_tag:
            # Need a consumer tag to register consumer info before sending
            # request to broker, because I/O might dispatch incoming messages
            # immediately following Basic.Consume-ok before _flush_output
            # returns
            consumer_tag = self._impl._generate_consumer_tag()

        if consumer_tag in self._consumer_infos:
            raise exceptions.DuplicateConsumerTag(consumer_tag)

        # Create new consumer
        self._consumer_infos[consumer_tag] = _ConsumerInfo(
            consumer_tag,
            no_ack=no_ack,
            consumer_cb=consumer_callback,
            alternate_event_sink=alternate_event_sink)

        try:
            with self._basic_consume_ok_result as ok_result:
                tag = self._impl.basic_consume(
                    consumer_callback=self._on_consumer_message_delivery,
                    queue=queue,
                    no_ack=no_ack,
                    exclusive=exclusive,
                    consumer_tag=consumer_tag,
                    arguments=arguments)

                assert tag == consumer_tag, (tag, consumer_tag)

                self._flush_output(ok_result.is_ready)
        except Exception:
            # If channel was closed, self._consumer_infos will be empty
            if consumer_tag in self._consumer_infos:
                del self._consumer_infos[consumer_tag]
            raise

        # NOTE: Consumer could get cancelled by broker immediately after opening
        # (e.g., queue getting deleted externally)
        if self._consumer_infos[consumer_tag].setting_up:
            self._consumer_infos[consumer_tag].state = _ConsumerInfo.ACTIVE

        return consumer_tag
List databases in the Impala cluster. Like the SHOW DATABASES command
dataset/ETHPy150Open pika/pika/pika/adapters/blocking_connection.py/BlockingChannel._basic_consume_impl--crossover with--dataset/ETHPy150Open cloudera/ibis/ibis/impala/client.py/ImpalaClient.list_databases