- if 'websocketUrl' in api._rootDesc:
- url = "{}?api_token={}".format(api._rootDesc['websocketUrl'], config.get('ARVADOS_API_TOKEN'))
- ws = EventClient(url, filters, on_event)
- ws.connect()
- else:
- _logger.info("Web sockets not available, falling back to log table polling")
- ws = PollClient(api, filters, on_event)
- return ws
- except Exception:
- if (ws):
- ws.close_connection()
- try:
- return PollClient(api, filters, on_event)
- except:
- raise
+ client.connect()
+ ok = True
+ return client
+ finally:
+ if not ok:
+ client.close_connection()
+
+def subscribe(api, filters, on_event, poll_fallback=15, last_log_id=None):
+ """
+ :api:
+ a client object retrieved from arvados.api(). The caller should not use this client object for anything else after calling subscribe().
+ :filters:
+ Initial subscription filters.
+ :on_event:
+ The callback when a message is received.
+ :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.
+ :last_log_id:
+ Log rows that are newer than the log id
+ """
+
+ if not poll_fallback:
+ return _subscribe_websocket(api, filters, on_event, last_log_id)
+
+ try:
+ return _subscribe_websocket(api, filters, on_event, last_log_id)
+ except Exception as e:
+ _logger.warn("Falling back to polling after websocket error: %s" % e)
+ p = PollClient(api, filters, on_event, poll_fallback, last_log_id)
+ p.start()
+ return p