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 import arvados
  11
  12 _logger = logging.getLogger('arvados.events')
  13
  14 class EventClient(WebSocketClient):
  15     def __init__(self, url, filters, on_event):
  16         ssl_options = None
  17         if re.match(r'(?i)^(true|1|yes)$',
  18                     config.get('ARVADOS_API_HOST_INSECURE', 'no')):
  19             ssl_options={'cert_reqs': ssl.CERT_NONE}
  20         else:
  21             ssl_options={'cert_reqs': ssl.CERT_REQUIRED}
  22         super(EventClient, self).__init__(url, ssl_options=ssl_options)
  23         self.filters = filters
  24         self.on_event = on_event
  25
  26     def opened(self):
  27         self.subscribe(self.filters)
  28
  29     def received_message(self, m):
  30         self.on_event(json.loads(str(m)))
  31
  32     def close_connection(self):
  33         try:
  34             self.sock.shutdown(socket.SHUT_RDWR)
  35             self.sock.close()
  36         except:
  37             pass
  38
  39     def subscribe(self, filters, last_log_id=None):
  40         m = {"method": "subscribe", "filters": filters}
  41         if last_log_id is not None:
  42             m["last_log_id"] = last_log_id
  43         self.send(json.dumps(m))
  44
  45     def unsubscribe(self, filters):
  46         self.send(json.dumps({"method": "unsubscribe", "filters": filters}))
  47
  48 class PollClient(threading.Thread):
  49     def __init__(self, api, filters, on_event, poll_time):
  50         super(PollClient, self).__init__()
  51         self.api = api
  52         if filters:
  53             self.filters = [filters]
  54         else:
  55             self.filters = []
  56         self.on_event = on_event
  57         self.poll_time = poll_time
  58         self.stop = threading.Event()
  59
  60     def run(self):
  61         items = self.api.logs().list(limit=1, order="id desc", filters=self.filters[0]).execute()['items']
  62         if len(items) > 0:
  63             self.id = items[0]["id"]
  64         else:
  65             self.id = 0
  66
  67         self.on_event({'status': 200})
  68
  69         while not self.stop.isSet():
  70             max_id = self.id
  71             for f in self.filters:
  72                 items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()['items']
  73                 for i in items:
  74                     if i['id'] > max_id:
  75                         max_id = i['id']
  76                     self.on_event(i)
  77             self.id = max_id
  78             self.stop.wait(self.poll_time)
  79
  80     def close(self):
  81         self.stop.set()
  82         self.join()
  83
  84     def subscribe(self, filters):
  85         self.on_event({'status': 200})
  86         self.filters.append(filters)
  87
  88     def unsubscribe(self, filters):
  89         del self.filters[self.filters.index(filters)]
  90
  91
  92 def subscribe(api, filters, on_event, poll_fallback=15):
  93     '''
  94     api: Must be a newly created from arvados.api(cache=False), not shared with the caller, as it may be used by a background thread.
  95     filters: Initial subscription filters.
  96     on_event: The callback when a message is received
  97     poll_fallback: If websockets are not available, fall back to polling every N seconds.  If poll_fallback=False, this will return None if websockets are not available.
  98     '''
  99     ws = None
 100     if 'websocketUrl' in api._rootDesc:
 101         try:
 102             url = "{}?api_token={}".format(api._rootDesc['websocketUrl'], api.api_token)
 103             ws = EventClient(url, filters, on_event)
 104             ws.connect()
 105             return ws
 106         except Exception as e:
 107             _logger.warn("Got exception %s trying to connect to websockets at %s" % (e, api._rootDesc['websocketUrl']))
 108             if ws:
 109                 ws.close_connection()
 110     if poll_fallback:
 111         _logger.warn("Websockets not available, falling back to log table polling")
 112         p = PollClient(api, filters, on_event, poll_fallback)
 113         p.start()
 114         return p
 115     else:
 116         _logger.error("Websockets not available")
 117         return None