mattermost/ws.py

   1 #!/usr/bin/env python3
   2 """
   3 Someone's Mattermost API v4 bindings.
   4   Copyright (c) 2016-2021 by Someone <someone@somenet.org> (aka. Jan Vales <jan@jvales.net>)
   5   published under MIT-License
   6 """
   7
   8 import logging
   9 import json
  10 import asyncio
  11 import threading
  12 import traceback
  13 import websockets
  14
  15 logger = logging.getLogger("mattermost.ws")
  16
  17
  18 class MMws:
  19     """
  20     Websocket client.
  21     """
  22
  23     def __init__(self, ws_handler, api, ws_url):
  24         self.api = api
  25         self.ws_url = ws_url
  26         self.ws_handler = ws_handler
  27         self.loop = asyncio.new_event_loop()
  28
  29         self._new_websocket_thread()
  30
  31
  32     def close_websocket(self):
  33         """
  34         Close the websocket and invalidate this object.
  35         """
  36         self.loop.stop()
  37         self.loop = None
  38
  39
  40     def _new_websocket_thread(self):
  41         if self.loop is None:
  42             return
  43
  44         thread = threading.Timer(3.0, self._open_websocket)
  45         thread.setDaemon(True)
  46         thread.setName("websocket")
  47         thread.start()
  48
  49
  50     def _open_websocket(self):
  51         try:
  52             self.loop.run_until_complete(self._websocket_run())
  53         except:
  54             if self.loop:
  55                 logger.info("websocket failed. restarting...")
  56                 logger.error("".join(traceback.format_exc()))
  57
  58         self._new_websocket_thread()
  59
  60
  61     async def _websocket_run(self):
  62         logger.info("Starting websocket client.")
  63         async with websockets.connect(self.ws_url, ping_interval=None, extra_headers={"Authorization": "Bearer "+self.api._bearer}) as websocket:
  64             #await websocket.send(json.dumps({"seq": 1, "action":"authentication_challenge", "data":{"token":self.api._bearer}}))
  65             logger.info(json.loads(await websocket.recv()))
  66             logger.info("websocket client connected. looping...")
  67
  68             while self.loop:
  69                 data = json.loads(await websocket.recv())
  70                 if "event" not in data:
  71                     continue
  72
  73                 try:
  74                     self.ws_handler(self, data)
  75                 except:
  76                     logger.error("".join(traceback.format_exc()))