12 from ws4py.client.threadedclient import WebSocketClient
14 _logger = logging.getLogger('arvados.events')
17 class EventClient(WebSocketClient):
18 def __init__(self, url, filters, on_event, last_log_id):
19 ssl_options = {'ca_certs': arvados.util.ca_certs_path()}
20 if config.flag_is_true('ARVADOS_API_HOST_INSECURE'):
21 ssl_options['cert_reqs'] = ssl.CERT_NONE
23 ssl_options['cert_reqs'] = ssl.CERT_REQUIRED
25 # Warning: If the host part of url resolves to both IPv6 and
26 # IPv4 addresses (common with "localhost"), only one of them
27 # will be attempted -- and it might not be the right one. See
28 # ws4py's WebSocketBaseClient.__init__.
29 super(EventClient, self).__init__(url, ssl_options=ssl_options)
30 self.filters = filters
31 self.on_event = on_event
32 self.last_log_id = last_log_id
33 self._closing_lock = threading.RLock()
35 self._closed = threading.Event()
38 self.subscribe(self.filters, self.last_log_id)
40 def closed(self, code, reason=None):
43 def received_message(self, m):
44 with self._closing_lock:
46 self.on_event(json.loads(str(m)))
48 def close(self, code=1000, reason='', timeout=0):
49 """Close event client and optionally wait for it to finish.
51 :timeout: is the number of seconds to wait for ws4py to
52 indicate that the connection has closed.
54 super(EventClient, self).close(code, reason)
55 with self._closing_lock:
56 # make sure we don't process any more messages.
58 # wait for ws4py to tell us the connection is closed.
59 self._closed.wait(timeout=timeout)
61 def subscribe(self, filters, last_log_id=None):
62 m = {"method": "subscribe", "filters": filters}
63 if last_log_id is not None:
64 m["last_log_id"] = last_log_id
65 self.send(json.dumps(m))
67 def unsubscribe(self, filters):
68 self.send(json.dumps({"method": "unsubscribe", "filters": filters}))
71 class PollClient(threading.Thread):
72 def __init__(self, api, filters, on_event, poll_time, last_log_id):
73 super(PollClient, self).__init__()
76 self.filters = [filters]
79 self.on_event = on_event
80 self.poll_time = poll_time
82 self.last_log_id = last_log_id
83 self._closing = threading.Event()
84 self._closing_lock = threading.RLock()
88 if self.last_log_id != None:
89 self.id = self.last_log_id
91 for f in self.filters:
92 items = self.api.logs().list(limit=1, order="id desc", filters=f).execute()['items']
94 if items[0]['id'] > self.id:
95 self.id = items[0]['id']
97 self.on_event({'status': 200})
99 while not self._closing.is_set():
102 for f in self.filters:
103 items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()
104 for i in items["items"]:
107 with self._closing_lock:
108 if self._closing.is_set():
111 if items["items_available"] > len(items["items"]):
115 self._closing.wait(self.poll_time)
117 def run_forever(self):
118 # Have to poll here, otherwise KeyboardInterrupt will never get processed.
119 while not self._closing.is_set():
120 self._closing.wait(1)
122 def close(self, code=None, reason=None, timeout=0):
123 """Close poll client and optionally wait for it to finish.
125 If an :on_event: handler is running in a different thread,
126 first wait (indefinitely) for it to return.
128 After closing, wait up to :timeout: seconds for the thread to
129 finish the poll request in progress (if any).
131 :code: and :reason: are ignored. They are present for
132 interface compatibility with EventClient.
135 with self._closing_lock:
138 self.join(timeout=timeout)
140 # "join() raises a RuntimeError if an attempt is made to join the
141 # current thread as that would cause a deadlock. It is also an
142 # error to join() a thread before it has been started and attempts
143 # to do so raises the same exception."
146 def subscribe(self, filters):
147 self.on_event({'status': 200})
148 self.filters.append(filters)
150 def unsubscribe(self, filters):
151 del self.filters[self.filters.index(filters)]
154 def _subscribe_websocket(api, filters, on_event, last_log_id=None):
155 endpoint = api._rootDesc.get('websocketUrl', None)
157 raise errors.FeatureNotEnabledError(
158 "Server does not advertise a websocket endpoint")
160 uri_with_token = "{}?api_token={}".format(endpoint, api.api_token)
161 client = EventClient(uri_with_token, filters, on_event, last_log_id)
169 client.close_connection()
171 _logger.warn("Failed to connect to websockets on %s" % endpoint)
175 def subscribe(api, filters, on_event, poll_fallback=15, last_log_id=None):
178 a client object retrieved from arvados.api(). The caller should not use this client object for anything else after calling subscribe().
180 Initial subscription filters.
182 The callback when a message is received.
184 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.
186 Log rows that are newer than the log id
189 if not poll_fallback:
190 return _subscribe_websocket(api, filters, on_event, last_log_id)
193 if not config.flag_is_true('ARVADOS_DISABLE_WEBSOCKETS'):
194 return _subscribe_websocket(api, filters, on_event, last_log_id)
196 _logger.info("Using polling because ARVADOS_DISABLE_WEBSOCKETS is true")
197 except Exception as e:
198 _logger.warn("Falling back to polling after websocket error: %s" % e)
199 p = PollClient(api, filters, on_event, poll_fallback, last_log_id)