code: def _connect_notification(self):
          
          try:
              self._ws = yield from self._session().ws_connect(self._getUrl("/notifications/ws"), auth=self._auth)
          except (aiohttp.WSServerHandshakeError, aiohttp.ClientResponseError):
              self._ws = None
          while self._ws is not None:
              try:
                  response = yield from self._ws.receive()
              except aiohttp.WSServerHandshakeError:
                  self._ws = None
                  break
              if response.tp == aiohttp.WSMsgType.closed or response.tp == aiohttp.WSMsgType.error or response.data is None:
                  self._connected = False
                  break
              msg = json.loads(response.data)
              action = msg.pop("action")
              event = msg.pop("event")
  
              if action == "ping":
                  self._cpu_usage_percent = event["cpu_usage_percent"]
                  self._memory_usage_percent = event["memory_usage_percent"]
                  self._controller.notification.emit("compute.updated", self.__json__())
              else:
                  yield from self._controller.notification.dispatch(action, event, compute_id=self.id)
          if self._ws:
              yield from self._ws.close()
  
          # Try to reconnect after 1 seconds if server unavailable only if not during tests (otherwise we create a ressources usage bomb)
          if not hasattr(sys, "_called_from_test") or not sys._called_from_test:
              asyncio.get_event_loop().call_later(1, lambda: asyncio.async(self.connect()))
          self._ws = None
          self._cpu_usage_percent = None
          self._memory_usage_percent = None
          self._controller.notification.emit("compute.updated", self.__json__())
docs: Connect to the notification stream