def consume(self, queue, no_ack=False,
                exclusive=False, arguments=None,
                inactivity_timeout=None):
        params = (queue, no_ack, exclusive)

        if self._queue_consumer_generator is not None:
            if params != self._queue_consumer_generator.params:
                raise ValueError(
                    'Consume with different params not allowed on existing '
                    'queue consumer generator; previous params: %r; '
                    'new params: %r'
                    % (self._queue_consumer_generator.params,
                       (queue, no_ack, exclusive)))
        else:
            LOGGER.debug('Creating new queue consumer generator; params: %r',
                         params)
            # Need a consumer tag to register consumer info before sending
            # request to broker, because I/O might pick up incoming messages
            # in addition to Basic.Consume-ok
            consumer_tag = self._impl._generate_consumer_tag()

            self._queue_consumer_generator = _QueueConsumerGeneratorInfo(
                params,
                consumer_tag)

            try:
                self._basic_consume_impl(
                    queue=queue,
                    no_ack=no_ack,
                    exclusive=exclusive,
                    consumer_tag=consumer_tag,
                    arguments=arguments,
                    alternate_event_sink=self._on_consumer_generator_event)
            except Exception:
                self._queue_consumer_generator = None
                raise

            LOGGER.info('Created new queue consumer generator %r',
                        self._queue_consumer_generator)

        while self._queue_consumer_generator is not None:
            if self._queue_consumer_generator.pending_events:
                evt = self._queue_consumer_generator.pending_events.popleft()
                if type(evt) is _ConsumerCancellationEvt:
                    # Consumer was cancelled by broker
                    self._queue_consumer_generator = None
                    break
                else:
                    yield (evt.method, evt.properties, evt.body)
                    continue

            # Wait for a message to arrive
            if inactivity_timeout is None:
                self.connection.process_data_events(time_limit=None)
                continue

            # Wait with inactivity timeout
            wait_start_time = time.time()
            wait_deadline = wait_start_time + inactivity_timeout
            delta = inactivity_timeout

            while (self._queue_consumer_generator is not None and
                   not self._queue_consumer_generator.pending_events):
                self.connection.process_data_events(time_limit=delta)

                if not self._queue_consumer_generator:
                    # Consumer was cancelled by client
                    break

                if self._queue_consumer_generator.pending_events:
                    # Got message(s)
                    break

                delta = wait_deadline - time.time()
                if delta <= 0.0:
                    # Signal inactivity timeout
                    yield None
                    break
Queries Google Compute Engine to see if the specified SSH key exists.
dataset/ETHPy150Open pika/pika/pika/adapters/blocking_connection.py/BlockingChannel.consume--crossover with--dataset/ETHPy150Open AppScale/appscale-tools/lib/agents/gce_agent.py/GCEAgent.does_ssh_key_exist