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()
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``.