code: def cancel(self):
          
          # Remove it from protocol and factory so it doesn't restart later.
          try:
              del self._protocol._consumers[self.queue]
          except (KeyError, AttributeError):
              pass
          try:
              del self._protocol.factory._consumers[self.queue]
          except (KeyError, AttributeError):
              pass
          # Signal to the _read loop it's time to stop and wait for it to finish
          # with whatever message it might be working on, then wait for the deferred
          # to fire which indicates it is done.
          self._running = False
          yield self._read_loop
          try:
              yield self._channel.basic_cancel(consumer_tag=self._tag)
          except pika.exceptions.AMQPChannelError:
              # Consumers are tied to channels, so if this channel is dead the
              # consumer should already be canceled (and we can't get to it anyway)
              pass
          try:
              yield self._channel.close()
          except pika.exceptions.AMQPChannelError:
              pass
          if not self.result.called:
              self.result.callback(self)
docs: Cancel the consumer and clean up resources associated with it.
          Consumers that are canceled are allowed to finish processing any
          messages before halting.
  
          Returns:
              defer.Deferred: A deferred that fires when the consumer has finished
              processing any message it was in the middle of and has been successfully
              canceled.