docs: async def run_consumer(timeout=None, dry_run=False):
      """Run the consumer until it finishes processing.
  
      :param timeout: Set maximum execution time before cancellation, or
          ``None`` (default) for unlimited.
      :param dry_run: If ``True``, don't actually dispatch messages, just
          dequeue them. Defaults to ``False``.
      """
      channel = state.MANAGER_CONTROL_CHANNEL
      scope = {
          'type': 'control_event',
          'channel': channel,
      }
  
      app = ApplicationCommunicator(ManagerConsumer, scope)
  
      channel_layer = get_channel_layer()
  
      async def _consume_loop():
          """Run a loop to consume messages off the channels layer."""
          while True:
              message = await channel_layer.receive(channel)
              if dry_run:
                  continue
              if message.get('type', {}) == '_resolwe_manager_quit':
                  break
              message.update(scope)
              await app.send_input(message)
  
      if timeout is None:
          await _consume_loop()
      try:
          # A further grace period to catch late messages.
          async with async_timeout.timeout(timeout or 1):
              await _consume_loop()
      except asyncio.TimeoutError:
          pass
  
      await app.wait()