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         self.id = 0
  62         for f in self.filters:
  63             items = self.api.logs().list(limit=1, order="id desc", filters=f).execute()['items']
  64             if items:
  65                 if items[0]['id'] > self.id:
  66                     self.id = items[0]['id']
  67
  68         self.on_event({'status': 200})
  69
  70         while not self.stop.isSet():
  71             max_id = self.id
  72             for f in self.filters:
  73                 items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()['items']
  74                 for i in items:
  75                     if i['id'] > max_id:
  76                         max_id = i['id']
  77                     self.on_event(i)
  78             self.id = max_id
  79             self.stop.wait(self.poll_time)
  80
  81     def run_forever(self):
  82         # Have to poll here, otherwise KeyboardInterrupt will never get processed.
  83         while not self.stop.is_set():
  84             self.stop.wait(1)
  85
  86     def close(self):
  87         self.stop.set()
  88         try:
  89             self.join()
  90         except RuntimeError:
  91             # "join() raises a RuntimeError if an attempt is made to join the
  92             # current thread as that would cause a deadlock. It is also an
  93             # error to join() a thread before it has been started and attempts
  94             # to do so raises the same exception."
  95             pass
  96
  97     def subscribe(self, filters):
  98         self.on_event({'status': 200})
  99         self.filters.append(filters)
 100
 101     def unsubscribe(self, filters):
 102         del self.filters[self.filters.index(filters)]
 103
 104
 105 def subscribe(api, filters, on_event, poll_fallback=15):
 106     '''
 107     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.
 108     filters: Initial subscription filters.
 109     on_event: The callback when a message is received
 110     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.
 111     '''
 112     ws = None
 113     if 'websocketUrl' in api._rootDesc:
 114         try:
 115             url = "{}?api_token={}".format(api._rootDesc['websocketUrl'], api.api_token)
 116             ws = EventClient(url, filters, on_event)
 117             ws.connect()
 118             return ws
 119         except Exception as e:
 120             _logger.warn("Got exception %s trying to connect to websockets at %s" % (e, api._rootDesc['websocketUrl']))
 121             if ws:
 122                 ws.close_connection()
 123     if poll_fallback:
 124         _logger.warn("Websockets not available, falling back to log table polling")
 125         p = PollClient(api, filters, on_event, poll_fallback)
 126         p.start()
 127         return p
 128     else:
 129         _logger.error("Websockets not available")
 130         return None