- 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:
- 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):
+ '''
+ 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.
+ '''
+ if not poll_fallback:
+ return _subscribe_websocket(api, filters, on_event)
+
+ try:
+ return _subscribe_websocket(api, filters, on_event)
+ except Exception as e:
+ _logger.warn("Falling back to polling after websocket error: %s" % e)
+ p = PollClient(api, filters, on_event, poll_fallback)
+ p.start()
+ return p