sdk/python/arvados/events.py

   1 from ws4py.client.threadedclient import WebSocketClient
   2 import threading
   3 import json
   4 import os
   5 import time
   6 import ssl
   7 import re
   8 import config
   9 import logging
  10
  11 _logger = logging.getLogger('arvados.events')
  12
  13 class EventClient(WebSocketClient):
  14     def __init__(self, url, filters, on_event):
  15         ssl_options = None
  16         if re.match(r'(?i)^(true|1|yes)$',
  17                     config.get('ARVADOS_API_HOST_INSECURE', 'no')):
  18             ssl_options={'cert_reqs': ssl.CERT_NONE}
  19         else:
  20             ssl_options={'cert_reqs': ssl.CERT_REQUIRED}
  21         super(EventClient, self).__init__(url, ssl_options=ssl_options)
  22         self.filters = filters
  23         self.on_event = on_event
  24
  25     def opened(self):
  26         self.subscribe(self.filters)
  27
  28     def received_message(self, m):
  29         self.on_event(json.loads(str(m)))
  30
  31     def close_connection(self):
  32         try:
  33             self.sock.shutdown(socket.SHUT_RDWR)
  34             self.sock.close()
  35         except:
  36             pass
  37
  38     def subscribe(self, filters, last_log_id=None):
  39         m = {"method": "subscribe", "filters": filters}
  40         if last_log_id is not None:
  41             m["last_log_id"] = last_log_id
  42         self.send(json.dumps(m))
  43
  44     def unsubscribe(self, filters):
  45         self.send(json.dumps({"method": "unsubscribe", "filters": filters}))
  46
  47 class PollClient(threading.Thread):
  48     def __init__(self, api, filters, on_event, poll_time):
  49         super(PollClient, self).__init__()
  50         self.api = api
  51         self.filters = [filters]
  52         self.on_event = on_event
  53         items = self.api.logs().list(limit=1, order="id desc", filters=filters).execute()['items']
  54         if len(items) > 0:
  55             self.id = items[0]["id"]
  56         else:
  57             self.id = 0
  58         self.poll_time = poll_time
  59         self.stop = threading.Event()
  60         self.on_event({'status': 200})
  61
  62     def run(self):
  63         while not self.stop.isSet():
  64             max_id = 0
  65             for f in self.filters:
  66                 items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()['items']
  67                 for i in items:
  68                     if i['id'] > max_id:
  69                         max_id = i['id']
  70                     self.on_event(i)
  71             self.id = max_id
  72             self.stop.wait(self.poll_time)
  73
  74     def close(self):
  75         self.stop.set()
  76         self.join()
  77
  78     def subscribe(self, filters):
  79         self.on_event({'status': 200})
  80         self.filters.append(filters)
  81
  82     def unsubscribe(self, filters):
  83         del self.filters[self.filters.index(filters)]
  84
  85 def subscribe(api, filters, on_event, poll_fallback=15):
  86     ws = None
  87     if 'websocketUrl' in api._rootDesc:
  88         try:
  89             url = "{}?api_token={}".format(api._rootDesc['websocketUrl'], api.api_token)
  90             ws = EventClient(url, filters, on_event)
  91             ws.connect()
  92             return ws
  93         except Exception as e:
  94             _logger.warn("Got exception %s trying to connect to web sockets at %s" % (e, api._rootDesc['websocketUrl']))
  95             if ws:
  96                 ws.close_connection()
  97     if poll_fallback:
  98         _logger.warn("Web sockets not available, falling back to log table polling")
  99         p = PollClient(api, filters, on_event, poll_fallback)
 100         p.start()
 101         return p
 102     else:
 103         _logger.error("Web sockets not available")
 104         return None