'The low-level implementation used by `basic_consume` and `consume`.
See `basic_consume` docstring for more info.
NOTE: exactly one of consumer_callback/alternate_event_sink musts be
non-None.
This method has one additional parameter alternate_event_sink over the
args described in `basic_consume`.
:param callable alternate_event_sink: if specified, _ConsumerDeliveryEvt
and _ConsumerCancellationEvt objects will be diverted to this
callback instead of being deposited in the channel\'s
`_pending_events` container. Signature:
alternate_event_sink(evt)
:raises pika.exceptions.DuplicateConsumerTag: if consumer with given
consumer_tag is already present.'

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):
        consumer_tag = self._impl._generate_consumer_tag()
    if (consumer_tag in self._consumer_infos):
        raise exceptions.DuplicateConsumerTag(consumer_tag)
    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 (consumer_tag in self._consumer_infos):
            del self._consumer_infos[consumer_tag]
        raise
    if self._consumer_infos[consumer_tag].setting_up:
        self._consumer_infos[consumer_tag].state = _ConsumerInfo.ACTIVE
    return consumer_tag
