code: def consume(self, timeout=None, loop=None):
          
          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()
docs: Start consuming the stream
  
          :param timeout: int: if it's given then it stops consumer after given
          number of seconds