<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def consume(self, timeout=None, loop=None):
 """Start consuming the stream :param timeout: int: if it's given then it stops consumer after given number of seconds """

        if self._consumer_fn is None:
            raise ValueError('Consumer function is not defined yet')

        logger.info('Start consuming the stream')

        @asyncio.coroutine
        def worker(conn_url):
            extra_headers = {
                'Connection': 'upgrade',
                'Upgrade': 'websocket',
                'Sec-Websocket-Version': 13,
            }

            ws = yield from websockets.connect(
                conn_url, extra_headers=extra_headers)

            if ws is None:
                raise RuntimeError("Couldn't connect to the '%s'" % conn_url)

            try:
                while True:
                    message = yield from ws.recv()
                    yield from self._consumer_fn(message)
            finally:
                yield from ws.close()

        if loop is None:
            loop = asyncio.new_event_loop()

        asyncio.set_event_loop(loop)
        try:
            task = worker(conn_url=self._conn_url)
            if timeout:
                logger.info('Running task with timeout %s sec', timeout)
                loop.run_until_complete(
                    asyncio.wait_for(task, timeout=timeout))
            else:
                loop.run_until_complete(task)
        except asyncio.TimeoutError:
            logger.info('Timeout is reached. Closing the loop')
            loop.close()
        except KeyboardInterrupt:
            logger.info('Closing the loop')
            loop.close()