title: Python: Websockets in Synchronous Program
body: <p>I have a bog-standard synchronous python program that needs to be able to read data from websockets and update the GUI with the data. However, asyncio creep is constantly tripping me up.</p>
  
  <p>How do I make a module that:</p>
  
  <ol>
  <li>accepts multiple subscriptions to multiple sources </li>
  <li>sends an update to the requester whenever there's data</li>
  <li>opens exactly one websocket connection per URL </li>
  <li>resets the websocket if it closes</li>
  </ol>
  
  <p>Here's what I have already, but it's failing at many points:</p>
  
  <ol>
  <li><code>run_forever()</code> means that the loop gets stuck before the subscription completes and then <code>handle()</code> is stuck in the falsey <code>while</code> loop</li>
  <li>it does not seem to want to restart sockets when they're down because a websockets object does not have a <code>connected</code> property (websocket without an s does, but I'm not clear on the differences and can't find info online either)</li>
  <li>I'm absolutely not sure if my approach is remotely correct.</li>
  </ol>
  
  <p>Been fighting with this for weeks. Would appreciate some pointers.</p>
  
  <pre><code>class WSClient():
      subscriptions = set()
      connections = {}
      started = False
  
      def __init__(self):
          self.loop = asyncio.get_event_loop()
  
      def start(self):
          self.started = True
          self.loop.run_until_complete(self.handle())
          self.loop.run_until_forever()  # problematic, because it does not allow new subscribe() events
  
      async def handle(self):
          while len(self.connections) &gt; 0:
              # listen to every websocket
              futures = [self.listen(self.connections[url]) for url in self.connections]
              done, pending = await asyncio.wait(futures)
  
              # the following is apparently necessary to avoid warnings
              # about non-retrieved exceptions etc
              try:
                  data, ws = done.pop().result()
              except Exception as e:
                  print("OTHER EXCEPTION", e)
  
              for task in pending:
                  task.cancel()
  
      async def listen(self, ws):
          try:
              async for data in ws:
                  data = json.loads(data)
                  # call the subscriber (listener) back when there's data
                  [s.listener._handle_result(data) for s in self.subscriptions if s.ws == ws]
          except Exception as e:
              print('ERROR LISTENING; RESTARTING SOCKET', e)
              await asyncio.sleep(2)
              self.restart_socket(ws)
  
      def subscribe(self, subscription):
          task = self.loop.create_task(self._subscribe(subscription))
          asyncio.gather(task)
  
          if not self.started:
              self.start()
  
      async def _subscribe(self, subscription):
          try:
              ws = self.connections.get(subscription.url, await websockets.connect(subscription.url))
              await ws.send(json.dumps(subscription.sub_msg))
  
              subscription.ws = ws
              self.connections[subscription.url] = ws
              self.subscriptions.add(subscription)
          except Exception as e:
              print("ERROR SUBSCRIBING; RETRYING", e)
              await asyncio.sleep(2)
              self.subscribe(subscription)
  
      def restart_socket(self, ws):
          for s in self.subscriptions:
              if s.ws == ws and not s.ws.connected:
                  print(s)
                  del self.connections[s.url]
                  self.subscribe(s)
  </code></pre>