<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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()