ringcentral.websocket.web_socket_client 

  1#!/usr/bin/env python
  2# encoding: utf-8
  3from observable import Observable
  4import websockets
  5from .web_socket_subscription import WebSocketSubscription
  6from .events import WebSocketEvents
  7import json
  8import asyncio
  9import uuid
 10
 11class WebSocketClient(Observable):
 12    def __init__(self, platform):
 13        Observable.__init__(self)
 14        self._platform = platform
 15        self._web_socket = None
 16        self._done = False
 17        self._is_ready = False
 18        self._heartbeat_task = None
 19        self._send_attempt_counter = 0
 20        self._subscription = None
 21
 22    async def create_new_connection(self):
 23        """
 24        Creates a new WebSocket connection.
 25
 26        Returns:
 27            Any: Response object containing the result of the connection creation.
 28
 29        Raises:
 30            Exception: If any error occurs during the process.
 31
 32        Note:
 33            - Retrieves the WebSocket token using `get_web_socket_token`.
 34            - Attempts to open a WebSocket connection using the retrieved token's URI and access token.
 35            - Triggers the createConnectionError event if an error occurs and raises the exception.
 36        """
 37        try:
 38            web_socket_token = self.get_web_socket_token()
 39            open_connection_response = await self.open_connection(
 40                web_socket_token["uri"], web_socket_token["ws_access_token"]
 41            )
 42            return open_connection_response
 43        except Exception as e:
 44            self.trigger(WebSocketEvents.createConnectionError, e)
 45            raise
 46
 47    def get_web_socket_token(self):
 48        """
 49            Retrieves a WebSocket token.
 50
 51            Returns:
 52                dict: WebSocket token containing URI and access token.
 53
 54            Raises:
 55                Exception: If any error occurs during the process.
 56
 57            Note:
 58                - Sends a POST request to the '/restapi/oauth/wstoken' endpoint to obtain the WebSocket token.
 59                - Returns the WebSocket token as a dictionary containing the URI and access token.
 60                - Triggers the getTokenError event if an error occurs and raises the exception.
 61        """
 62        try:
 63            response = self._platform.post("/restapi/oauth/wstoken", body={})
 64            return response.json_dict()
 65        except Exception as e:
 66            self.trigger(WebSocketEvents.getTokenError, e)
 67            raise
 68
 69    async def open_connection(self, ws_uri, ws_access_token):
 70        """
 71            Opens a WebSocket connection.
 72
 73            Args:
 74                ws_uri (str): The WebSocket URI.
 75                ws_access_token (str): The access token for WebSocket authentication.
 76
 77            Raises:
 78                Exception: If any error occurs during the process.
 79
 80            Note:
 81                - Attempts to establish a WebSocket connection to the provided URI with the given access token.
 82                - Upon successful connection, sets up a heartbeat mechanism to maintain the connection.
 83                - Triggers the connectionCreated event upon successful connection establishment.
 84                - Listens for incoming messages and triggers the receiveMessage event for each received message.
 85                - Isolates receive-message handler failures so a handler that raises does not stop delivery to the remaining handlers or reception of future messages; each failure triggers the receiveMessageError event with the original exception.
 86                - Triggers the createConnectionError event if an error occurs while establishing the connection or during the initial handshake, and raises the exception.
 87                - A failure from receiving messages after the initial handshake triggers the receiveMessageError event once; it is not reported as a connection-creation error and does not propagate to the connection-setup or recovery wrappers.
 88                - When the receive loop terminates (receive failure, cancellation, or intentional closure), the client is marked not ready and its heartbeat task is cancelled.
 89                - An intentional closure does not trigger the receiveMessageError event.
 90        """
 91        self._done = False
 92        try:
 93            websocket = await websockets.connect(
 94                f"{ws_uri}?access_token={ws_access_token}"
 95            )
 96            connectionMessage = await websocket.recv()
 97        except Exception as e:
 98            self.trigger(WebSocketEvents.createConnectionError, e)
 99            raise
100
101        connection_info = {}
102        connection_info["connection"] = websocket
103        connection_info["connection_details"] = connectionMessage
104        self._web_socket = connection_info
105        self._is_ready = True
106        self.trigger(WebSocketEvents.connectionCreated, self)
107
108        # heartbeat every 10 minutes
109        async def timer_function():
110            while True:
111                if self._done:
112                    timer.cancel()
113                    break
114                await asyncio.sleep(600)
115                await self.send_message([{"type": "Heartbeat", "messageId": str(uuid.uuid4())}])
116        timer = asyncio.create_task(timer_function())
117        self._heartbeat_task = timer
118
119        try:
120            await asyncio.sleep(0)
121            while True:
122                message = await websocket.recv()
123                self.trigger(WebSocketEvents.receiveMessage, message)
124                await asyncio.sleep(0)
125        except Exception as e:
126            if not self._done:
127                self.trigger(WebSocketEvents.receiveMessageError, e)
128        finally:
129            self._is_ready = False
130            timer.cancel()
131
132    def trigger(self, event, *args, **kw):
133        """
134            Triggers the handlers registered for an event.
135
136            Args:
137                event (str): The event to trigger.
138                args: The event arguments.
139                kw: The event keyword arguments.
140
141            Returns:
142                bool: True if any handler was triggered, False if the event has no handlers.
143
144            Note:
145                - receiveMessage handlers are invoked independently: if one raises, the remaining handlers still receive the same raw message and one receiveMessageError event is triggered with the original exception.
146                - receiveMessageError handlers are also invoked independently: a failing error handler is contained so error reporting and reception continue.
147                - Every other event keeps the default dispatch behavior.
148        """
149        if event == WebSocketEvents.receiveMessage:
150            return self._trigger_receive_message(*args, **kw)
151        if event == WebSocketEvents.receiveMessageError:
152            return self._trigger_receive_message_error(*args, **kw)
153        return Observable.trigger(self, event, *args, **kw)
154
155    def _trigger_receive_message(self, *args, **kw):
156        handlers = list(self.events.get(WebSocketEvents.receiveMessage) or [])
157        if not handlers:
158            return False
159        for handler in handlers:
160            try:
161                handler(*args, **kw)
162            except Exception as e:
163                self.trigger(WebSocketEvents.receiveMessageError, e)
164        return True
165
166    def _trigger_receive_message_error(self, *args, **kw):
167        handlers = list(self.events.get(WebSocketEvents.receiveMessageError) or [])
168        if not handlers:
169            return False
170        for handler in handlers:
171            try:
172                handler(*args, **kw)
173            except Exception:
174                pass
175        return True
176
177    def get_connection_info(self):
178        return self._web_socket
179
180    def get_connection(self):
181        return self._web_socket["connection"]
182
183    async def close_connection(self):
184        """
185            Closes the WebSocket connection.
186
187            Raises:
188                Exception: If any error occurs during the process.
189
190            Note:
191                - Sets the `_done` flag to True to signal the termination of the heartbeat mechanism.
192                - Sets the `_is_ready` flag to False to indicate that the connection is no longer ready.
193                - Retrieves the WebSocket connection using `get_connection`.
194                - Closes the WebSocket connection.
195                - Triggers the closeConnectionError event if an error occurs during the closing process and raises the exception.
196        """
197        try:
198            self._done = True
199            self._is_ready = False
200            ws_connection = self.get_connection()
201            await ws_connection.close()
202        except Exception as e:
203            self.trigger(WebSocketEvents.closeConnectionError, e)
204            raise
205
206    async def recover_connection(self):
207        try:
208            ws_connection_info = self.get_web_socket_token()
209            recovered_connection_info = await self.open_connection(
210                ws_connection_info["uri"], ws_connection_info["ws_access_token"]
211            )
212            # if recovered_connection_info['recoveryState'] === "Successful", then subscription is restored
213            # otherwise, need to create a new subscription
214            # IMPORTANT: WebSocket creation is successful if it doesn't raise any exception
215            return recovered_connection_info
216        except Exception as e:
217            self.trigger(WebSocketEvents.recoverConnectionError, e)
218            raise
219
220    async def send_message(self, message):
221        """
222            Sends a message over the WebSocket connection.
223
224            Args:
225                message (Any): The message to be sent.
226
227            Raises:
228                Exception: If any error occurs during the process or if the connection is not ready after multiple attempts.
229
230            Note:
231                - Checks if the WebSocket connection is ready (`_is_ready` flag).
232                - If the connection is ready, resets the send attempt counter and sends the message.
233                - If the connection is not ready, retries after a delay and increments the send attempt counter.
234                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
235        """
236        try:
237            if self._is_ready:
238                self._send_attempt_counter = 0
239                requestBodyJson = json.dumps(message)
240                await self.get_connection().send(requestBodyJson)
241            else:
242                await asyncio.sleep(1)
243                await self.send_message(message)
244                self._send_attempt_counter += 1
245                if(self._send_attempt_counter > 10):
246                    self.trigger(WebSocketEvents.connectionNotReady)
247                    self._send_attempt_counter = 0
248                    raise
249        except Exception as e:
250            self.trigger(WebSocketEvents.sendMessageError, e)
251            raise
252
253    async def create_subscription(self, events):
254        """
255            Creates a subscription to WebSocket events.
256
257            Args:
258                events (list): A required, non-empty list of events to subscribe to.
259
260            Raises:
261                Exception: If any error occurs during the process, if events are omitted, None, or empty, if a subscription creation is already in progress, or if a subscription already exists.
262
263            Note:
264                - Events are required: an omitted argument is rejected by ordinary Python argument checking, and None or empty events raise "Events are undefined" before any request is sent or any receive listener is attached.
265                - The client retains and reuses a single subscription object; a retry after a failed creation and a creation after removal reuse the retained object's existing registration flow with the newly supplied events.
266                - Rejects a call while a subscription creation is already in progress with "WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying".
267                - Rejects a call after a subscription has been created with "A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another".
268                - Rejections are delivered through the createSubscriptionError event and raised to the caller.
269                - If the WebSocket connection is ready (`_is_ready` flag), resets the send attempt counter and registers the retained subscription with the specified events.
270                - If the connection is not ready, retries after a delay and increments the send attempt counter.
271                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
272        """
273        try:
274            if not events or len(events) == 0:
275                raise Exception("Events are undefined")
276
277            if self._subscription is None:
278                self._subscription = WebSocketSubscription(self)
279            subscription = self._subscription
280
281            if subscription._pending_creation_message_id is not None:
282                raise Exception("WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying")
283            if subscription.get_subscription_info() is not None:
284                raise Exception("A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another")
285
286            if self._is_ready:
287                self._send_attempt_counter = 0
288                await subscription.register(events)
289            else:
290                await asyncio.sleep(1)
291                await self.create_subscription(events)
292                self._send_attempt_counter += 1
293                if(self._send_attempt_counter > 10):
294                    self.trigger(WebSocketEvents.connectionNotReady)
295                    self._send_attempt_counter = 0
296                    raise
297
298        except Exception as e:
299            self.trigger(WebSocketEvents.createSubscriptionError, e)
300            raise
301
302    async def update_subscription(self, subscription, events=None):
303        """
304            Updates an existing WebSocket subscription with new events.
305
306            Args:
307                subscription : The WebSocket subscription to update.
308                events (list, optional): A list of events to update the subscription with. Default is None.
309
310            Returns:
311                WebSocketSubscription: The updated WebSocket subscription.
312
313            Raises:
314                Exception: If any error occurs during the process.
315
316            Note:
317                - Updates the specified WebSocket subscription with the new events provided.
318                - If the update is successful, returns the updated WebSocket subscription.
319                - If an error occurs during the update process, triggers the updateSubscriptionError event and raises an exception.
320        """
321        try:
322            await subscription.update(events)
323            return subscription
324        except Exception as e:
325            self.trigger(WebSocketEvents.updateSubscriptionError, e)
326            raise
327
328    async def remove_subscription(self, subscription):
329        """
330            Removes an existing WebSocket subscription.
331
332            Args:
333                subscription : The WebSocket subscription to remove.
334
335            Raises:
336                Exception: If any error occurs during the removal process.
337
338            Note:
339                - Removes the specified WebSocket subscription.
340                - If the removal is successful, the subscription is effectively unsubscribed from the events it was subscribed to.
341                - If an error occurs during the removal process, triggers the removeSubscriptionError event and raises an exception.
342        """
343        try:
344            await subscription.remove()
345        except Exception as e:
346            self.trigger(WebSocketEvents.removeSubscriptionError, e)
347            raise
348
349
350if __name__ == "__main__":
351    pass
class WebSocketClient(observable.core.Observable): 
 12class WebSocketClient(Observable):
 13    def __init__(self, platform):
 14        Observable.__init__(self)
 15        self._platform = platform
 16        self._web_socket = None
 17        self._done = False
 18        self._is_ready = False
 19        self._heartbeat_task = None
 20        self._send_attempt_counter = 0
 21        self._subscription = None
 22
 23    async def create_new_connection(self):
 24        """
 25        Creates a new WebSocket connection.
 26
 27        Returns:
 28            Any: Response object containing the result of the connection creation.
 29
 30        Raises:
 31            Exception: If any error occurs during the process.
 32
 33        Note:
 34            - Retrieves the WebSocket token using `get_web_socket_token`.
 35            - Attempts to open a WebSocket connection using the retrieved token's URI and access token.
 36            - Triggers the createConnectionError event if an error occurs and raises the exception.
 37        """
 38        try:
 39            web_socket_token = self.get_web_socket_token()
 40            open_connection_response = await self.open_connection(
 41                web_socket_token["uri"], web_socket_token["ws_access_token"]
 42            )
 43            return open_connection_response
 44        except Exception as e:
 45            self.trigger(WebSocketEvents.createConnectionError, e)
 46            raise
 47
 48    def get_web_socket_token(self):
 49        """
 50            Retrieves a WebSocket token.
 51
 52            Returns:
 53                dict: WebSocket token containing URI and access token.
 54
 55            Raises:
 56                Exception: If any error occurs during the process.
 57
 58            Note:
 59                - Sends a POST request to the '/restapi/oauth/wstoken' endpoint to obtain the WebSocket token.
 60                - Returns the WebSocket token as a dictionary containing the URI and access token.
 61                - Triggers the getTokenError event if an error occurs and raises the exception.
 62        """
 63        try:
 64            response = self._platform.post("/restapi/oauth/wstoken", body={})
 65            return response.json_dict()
 66        except Exception as e:
 67            self.trigger(WebSocketEvents.getTokenError, e)
 68            raise
 69
 70    async def open_connection(self, ws_uri, ws_access_token):
 71        """
 72            Opens a WebSocket connection.
 73
 74            Args:
 75                ws_uri (str): The WebSocket URI.
 76                ws_access_token (str): The access token for WebSocket authentication.
 77
 78            Raises:
 79                Exception: If any error occurs during the process.
 80
 81            Note:
 82                - Attempts to establish a WebSocket connection to the provided URI with the given access token.
 83                - Upon successful connection, sets up a heartbeat mechanism to maintain the connection.
 84                - Triggers the connectionCreated event upon successful connection establishment.
 85                - Listens for incoming messages and triggers the receiveMessage event for each received message.
 86                - Isolates receive-message handler failures so a handler that raises does not stop delivery to the remaining handlers or reception of future messages; each failure triggers the receiveMessageError event with the original exception.
 87                - Triggers the createConnectionError event if an error occurs while establishing the connection or during the initial handshake, and raises the exception.
 88                - A failure from receiving messages after the initial handshake triggers the receiveMessageError event once; it is not reported as a connection-creation error and does not propagate to the connection-setup or recovery wrappers.
 89                - When the receive loop terminates (receive failure, cancellation, or intentional closure), the client is marked not ready and its heartbeat task is cancelled.
 90                - An intentional closure does not trigger the receiveMessageError event.
 91        """
 92        self._done = False
 93        try:
 94            websocket = await websockets.connect(
 95                f"{ws_uri}?access_token={ws_access_token}"
 96            )
 97            connectionMessage = await websocket.recv()
 98        except Exception as e:
 99            self.trigger(WebSocketEvents.createConnectionError, e)
100            raise
101
102        connection_info = {}
103        connection_info["connection"] = websocket
104        connection_info["connection_details"] = connectionMessage
105        self._web_socket = connection_info
106        self._is_ready = True
107        self.trigger(WebSocketEvents.connectionCreated, self)
108
109        # heartbeat every 10 minutes
110        async def timer_function():
111            while True:
112                if self._done:
113                    timer.cancel()
114                    break
115                await asyncio.sleep(600)
116                await self.send_message([{"type": "Heartbeat", "messageId": str(uuid.uuid4())}])
117        timer = asyncio.create_task(timer_function())
118        self._heartbeat_task = timer
119
120        try:
121            await asyncio.sleep(0)
122            while True:
123                message = await websocket.recv()
124                self.trigger(WebSocketEvents.receiveMessage, message)
125                await asyncio.sleep(0)
126        except Exception as e:
127            if not self._done:
128                self.trigger(WebSocketEvents.receiveMessageError, e)
129        finally:
130            self._is_ready = False
131            timer.cancel()
132
133    def trigger(self, event, *args, **kw):
134        """
135            Triggers the handlers registered for an event.
136
137            Args:
138                event (str): The event to trigger.
139                args: The event arguments.
140                kw: The event keyword arguments.
141
142            Returns:
143                bool: True if any handler was triggered, False if the event has no handlers.
144
145            Note:
146                - receiveMessage handlers are invoked independently: if one raises, the remaining handlers still receive the same raw message and one receiveMessageError event is triggered with the original exception.
147                - receiveMessageError handlers are also invoked independently: a failing error handler is contained so error reporting and reception continue.
148                - Every other event keeps the default dispatch behavior.
149        """
150        if event == WebSocketEvents.receiveMessage:
151            return self._trigger_receive_message(*args, **kw)
152        if event == WebSocketEvents.receiveMessageError:
153            return self._trigger_receive_message_error(*args, **kw)
154        return Observable.trigger(self, event, *args, **kw)
155
156    def _trigger_receive_message(self, *args, **kw):
157        handlers = list(self.events.get(WebSocketEvents.receiveMessage) or [])
158        if not handlers:
159            return False
160        for handler in handlers:
161            try:
162                handler(*args, **kw)
163            except Exception as e:
164                self.trigger(WebSocketEvents.receiveMessageError, e)
165        return True
166
167    def _trigger_receive_message_error(self, *args, **kw):
168        handlers = list(self.events.get(WebSocketEvents.receiveMessageError) or [])
169        if not handlers:
170            return False
171        for handler in handlers:
172            try:
173                handler(*args, **kw)
174            except Exception:
175                pass
176        return True
177
178    def get_connection_info(self):
179        return self._web_socket
180
181    def get_connection(self):
182        return self._web_socket["connection"]
183
184    async def close_connection(self):
185        """
186            Closes the WebSocket connection.
187
188            Raises:
189                Exception: If any error occurs during the process.
190
191            Note:
192                - Sets the `_done` flag to True to signal the termination of the heartbeat mechanism.
193                - Sets the `_is_ready` flag to False to indicate that the connection is no longer ready.
194                - Retrieves the WebSocket connection using `get_connection`.
195                - Closes the WebSocket connection.
196                - Triggers the closeConnectionError event if an error occurs during the closing process and raises the exception.
197        """
198        try:
199            self._done = True
200            self._is_ready = False
201            ws_connection = self.get_connection()
202            await ws_connection.close()
203        except Exception as e:
204            self.trigger(WebSocketEvents.closeConnectionError, e)
205            raise
206
207    async def recover_connection(self):
208        try:
209            ws_connection_info = self.get_web_socket_token()
210            recovered_connection_info = await self.open_connection(
211                ws_connection_info["uri"], ws_connection_info["ws_access_token"]
212            )
213            # if recovered_connection_info['recoveryState'] === "Successful", then subscription is restored
214            # otherwise, need to create a new subscription
215            # IMPORTANT: WebSocket creation is successful if it doesn't raise any exception
216            return recovered_connection_info
217        except Exception as e:
218            self.trigger(WebSocketEvents.recoverConnectionError, e)
219            raise
220
221    async def send_message(self, message):
222        """
223            Sends a message over the WebSocket connection.
224
225            Args:
226                message (Any): The message to be sent.
227
228            Raises:
229                Exception: If any error occurs during the process or if the connection is not ready after multiple attempts.
230
231            Note:
232                - Checks if the WebSocket connection is ready (`_is_ready` flag).
233                - If the connection is ready, resets the send attempt counter and sends the message.
234                - If the connection is not ready, retries after a delay and increments the send attempt counter.
235                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
236        """
237        try:
238            if self._is_ready:
239                self._send_attempt_counter = 0
240                requestBodyJson = json.dumps(message)
241                await self.get_connection().send(requestBodyJson)
242            else:
243                await asyncio.sleep(1)
244                await self.send_message(message)
245                self._send_attempt_counter += 1
246                if(self._send_attempt_counter > 10):
247                    self.trigger(WebSocketEvents.connectionNotReady)
248                    self._send_attempt_counter = 0
249                    raise
250        except Exception as e:
251            self.trigger(WebSocketEvents.sendMessageError, e)
252            raise
253
254    async def create_subscription(self, events):
255        """
256            Creates a subscription to WebSocket events.
257
258            Args:
259                events (list): A required, non-empty list of events to subscribe to.
260
261            Raises:
262                Exception: If any error occurs during the process, if events are omitted, None, or empty, if a subscription creation is already in progress, or if a subscription already exists.
263
264            Note:
265                - Events are required: an omitted argument is rejected by ordinary Python argument checking, and None or empty events raise "Events are undefined" before any request is sent or any receive listener is attached.
266                - The client retains and reuses a single subscription object; a retry after a failed creation and a creation after removal reuse the retained object's existing registration flow with the newly supplied events.
267                - Rejects a call while a subscription creation is already in progress with "WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying".
268                - Rejects a call after a subscription has been created with "A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another".
269                - Rejections are delivered through the createSubscriptionError event and raised to the caller.
270                - If the WebSocket connection is ready (`_is_ready` flag), resets the send attempt counter and registers the retained subscription with the specified events.
271                - If the connection is not ready, retries after a delay and increments the send attempt counter.
272                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
273        """
274        try:
275            if not events or len(events) == 0:
276                raise Exception("Events are undefined")
277
278            if self._subscription is None:
279                self._subscription = WebSocketSubscription(self)
280            subscription = self._subscription
281
282            if subscription._pending_creation_message_id is not None:
283                raise Exception("WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying")
284            if subscription.get_subscription_info() is not None:
285                raise Exception("A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another")
286
287            if self._is_ready:
288                self._send_attempt_counter = 0
289                await subscription.register(events)
290            else:
291                await asyncio.sleep(1)
292                await self.create_subscription(events)
293                self._send_attempt_counter += 1
294                if(self._send_attempt_counter > 10):
295                    self.trigger(WebSocketEvents.connectionNotReady)
296                    self._send_attempt_counter = 0
297                    raise
298
299        except Exception as e:
300            self.trigger(WebSocketEvents.createSubscriptionError, e)
301            raise
302
303    async def update_subscription(self, subscription, events=None):
304        """
305            Updates an existing WebSocket subscription with new events.
306
307            Args:
308                subscription : The WebSocket subscription to update.
309                events (list, optional): A list of events to update the subscription with. Default is None.
310
311            Returns:
312                WebSocketSubscription: The updated WebSocket subscription.
313
314            Raises:
315                Exception: If any error occurs during the process.
316
317            Note:
318                - Updates the specified WebSocket subscription with the new events provided.
319                - If the update is successful, returns the updated WebSocket subscription.
320                - If an error occurs during the update process, triggers the updateSubscriptionError event and raises an exception.
321        """
322        try:
323            await subscription.update(events)
324            return subscription
325        except Exception as e:
326            self.trigger(WebSocketEvents.updateSubscriptionError, e)
327            raise
328
329    async def remove_subscription(self, subscription):
330        """
331            Removes an existing WebSocket subscription.
332
333            Args:
334                subscription : The WebSocket subscription to remove.
335
336            Raises:
337                Exception: If any error occurs during the removal process.
338
339            Note:
340                - Removes the specified WebSocket subscription.
341                - If the removal is successful, the subscription is effectively unsubscribed from the events it was subscribed to.
342                - If an error occurs during the removal process, triggers the removeSubscriptionError event and raises an exception.
343        """
344        try:
345            await subscription.remove()
346        except Exception as e:
347            self.trigger(WebSocketEvents.removeSubscriptionError, e)
348            raise

Event system for python

WebSocketClient(platform) 
13    def __init__(self, platform):
14        Observable.__init__(self)
15        self._platform = platform
16        self._web_socket = None
17        self._done = False
18        self._is_ready = False
19        self._heartbeat_task = None
20        self._send_attempt_counter = 0
21        self._subscription = None
async def create_new_connection(self): 
23    async def create_new_connection(self):
24        """
25        Creates a new WebSocket connection.
26
27        Returns:
28            Any: Response object containing the result of the connection creation.
29
30        Raises:
31            Exception: If any error occurs during the process.
32
33        Note:
34            - Retrieves the WebSocket token using `get_web_socket_token`.
35            - Attempts to open a WebSocket connection using the retrieved token's URI and access token.
36            - Triggers the createConnectionError event if an error occurs and raises the exception.
37        """
38        try:
39            web_socket_token = self.get_web_socket_token()
40            open_connection_response = await self.open_connection(
41                web_socket_token["uri"], web_socket_token["ws_access_token"]
42            )
43            return open_connection_response
44        except Exception as e:
45            self.trigger(WebSocketEvents.createConnectionError, e)
46            raise

Creates a new WebSocket connection.

Returns: Any: Response object containing the result of the connection creation.

Raises: Exception: If any error occurs during the process.

Note: - Retrieves the WebSocket token using get_web_socket_token. - Attempts to open a WebSocket connection using the retrieved token's URI and access token. - Triggers the createConnectionError event if an error occurs and raises the exception.

def get_web_socket_token(self): 
48    def get_web_socket_token(self):
49        """
50            Retrieves a WebSocket token.
51
52            Returns:
53                dict: WebSocket token containing URI and access token.
54
55            Raises:
56                Exception: If any error occurs during the process.
57
58            Note:
59                - Sends a POST request to the '/restapi/oauth/wstoken' endpoint to obtain the WebSocket token.
60                - Returns the WebSocket token as a dictionary containing the URI and access token.
61                - Triggers the getTokenError event if an error occurs and raises the exception.
62        """
63        try:
64            response = self._platform.post("/restapi/oauth/wstoken", body={})
65            return response.json_dict()
66        except Exception as e:
67            self.trigger(WebSocketEvents.getTokenError, e)
68            raise

Retrieves a WebSocket token.

Returns: dict: WebSocket token containing URI and access token.

Raises: Exception: If any error occurs during the process.

Note: - Sends a POST request to the '/restapi/oauth/wstoken' endpoint to obtain the WebSocket token. - Returns the WebSocket token as a dictionary containing the URI and access token. - Triggers the getTokenError event if an error occurs and raises the exception.

async def open_connection(self, ws_uri, ws_access_token): 
 70    async def open_connection(self, ws_uri, ws_access_token):
 71        """
 72            Opens a WebSocket connection.
 73
 74            Args:
 75                ws_uri (str): The WebSocket URI.
 76                ws_access_token (str): The access token for WebSocket authentication.
 77
 78            Raises:
 79                Exception: If any error occurs during the process.
 80
 81            Note:
 82                - Attempts to establish a WebSocket connection to the provided URI with the given access token.
 83                - Upon successful connection, sets up a heartbeat mechanism to maintain the connection.
 84                - Triggers the connectionCreated event upon successful connection establishment.
 85                - Listens for incoming messages and triggers the receiveMessage event for each received message.
 86                - Isolates receive-message handler failures so a handler that raises does not stop delivery to the remaining handlers or reception of future messages; each failure triggers the receiveMessageError event with the original exception.
 87                - Triggers the createConnectionError event if an error occurs while establishing the connection or during the initial handshake, and raises the exception.
 88                - A failure from receiving messages after the initial handshake triggers the receiveMessageError event once; it is not reported as a connection-creation error and does not propagate to the connection-setup or recovery wrappers.
 89                - When the receive loop terminates (receive failure, cancellation, or intentional closure), the client is marked not ready and its heartbeat task is cancelled.
 90                - An intentional closure does not trigger the receiveMessageError event.
 91        """
 92        self._done = False
 93        try:
 94            websocket = await websockets.connect(
 95                f"{ws_uri}?access_token={ws_access_token}"
 96            )
 97            connectionMessage = await websocket.recv()
 98        except Exception as e:
 99            self.trigger(WebSocketEvents.createConnectionError, e)
100            raise
101
102        connection_info = {}
103        connection_info["connection"] = websocket
104        connection_info["connection_details"] = connectionMessage
105        self._web_socket = connection_info
106        self._is_ready = True
107        self.trigger(WebSocketEvents.connectionCreated, self)
108
109        # heartbeat every 10 minutes
110        async def timer_function():
111            while True:
112                if self._done:
113                    timer.cancel()
114                    break
115                await asyncio.sleep(600)
116                await self.send_message([{"type": "Heartbeat", "messageId": str(uuid.uuid4())}])
117        timer = asyncio.create_task(timer_function())
118        self._heartbeat_task = timer
119
120        try:
121            await asyncio.sleep(0)
122            while True:
123                message = await websocket.recv()
124                self.trigger(WebSocketEvents.receiveMessage, message)
125                await asyncio.sleep(0)
126        except Exception as e:
127            if not self._done:
128                self.trigger(WebSocketEvents.receiveMessageError, e)
129        finally:
130            self._is_ready = False
131            timer.cancel()

Opens a WebSocket connection.

Args: ws_uri (str): The WebSocket URI. ws_access_token (str): The access token for WebSocket authentication.

Raises: Exception: If any error occurs during the process.

Note: - Attempts to establish a WebSocket connection to the provided URI with the given access token. - Upon successful connection, sets up a heartbeat mechanism to maintain the connection. - Triggers the connectionCreated event upon successful connection establishment. - Listens for incoming messages and triggers the receiveMessage event for each received message. - Isolates receive-message handler failures so a handler that raises does not stop delivery to the remaining handlers or reception of future messages; each failure triggers the receiveMessageError event with the original exception. - Triggers the createConnectionError event if an error occurs while establishing the connection or during the initial handshake, and raises the exception. - A failure from receiving messages after the initial handshake triggers the receiveMessageError event once; it is not reported as a connection-creation error and does not propagate to the connection-setup or recovery wrappers. - When the receive loop terminates (receive failure, cancellation, or intentional closure), the client is marked not ready and its heartbeat task is cancelled. - An intentional closure does not trigger the receiveMessageError event.

def trigger(self, event, *args, **kw): 
133    def trigger(self, event, *args, **kw):
134        """
135            Triggers the handlers registered for an event.
136
137            Args:
138                event (str): The event to trigger.
139                args: The event arguments.
140                kw: The event keyword arguments.
141
142            Returns:
143                bool: True if any handler was triggered, False if the event has no handlers.
144
145            Note:
146                - receiveMessage handlers are invoked independently: if one raises, the remaining handlers still receive the same raw message and one receiveMessageError event is triggered with the original exception.
147                - receiveMessageError handlers are also invoked independently: a failing error handler is contained so error reporting and reception continue.
148                - Every other event keeps the default dispatch behavior.
149        """
150        if event == WebSocketEvents.receiveMessage:
151            return self._trigger_receive_message(*args, **kw)
152        if event == WebSocketEvents.receiveMessageError:
153            return self._trigger_receive_message_error(*args, **kw)
154        return Observable.trigger(self, event, *args, **kw)

Triggers the handlers registered for an event.

Args: event (str): The event to trigger. args: The event arguments. kw: The event keyword arguments.

Returns: bool: True if any handler was triggered, False if the event has no handlers.

Note: - receiveMessage handlers are invoked independently: if one raises, the remaining handlers still receive the same raw message and one receiveMessageError event is triggered with the original exception. - receiveMessageError handlers are also invoked independently: a failing error handler is contained so error reporting and reception continue. - Every other event keeps the default dispatch behavior.

def get_connection_info(self): 
178    def get_connection_info(self):
179        return self._web_socket
def get_connection(self): 
181    def get_connection(self):
182        return self._web_socket["connection"]
async def close_connection(self): 
184    async def close_connection(self):
185        """
186            Closes the WebSocket connection.
187
188            Raises:
189                Exception: If any error occurs during the process.
190
191            Note:
192                - Sets the `_done` flag to True to signal the termination of the heartbeat mechanism.
193                - Sets the `_is_ready` flag to False to indicate that the connection is no longer ready.
194                - Retrieves the WebSocket connection using `get_connection`.
195                - Closes the WebSocket connection.
196                - Triggers the closeConnectionError event if an error occurs during the closing process and raises the exception.
197        """
198        try:
199            self._done = True
200            self._is_ready = False
201            ws_connection = self.get_connection()
202            await ws_connection.close()
203        except Exception as e:
204            self.trigger(WebSocketEvents.closeConnectionError, e)
205            raise

Closes the WebSocket connection.

Raises: Exception: If any error occurs during the process.

Note: - Sets the _done flag to True to signal the termination of the heartbeat mechanism. - Sets the _is_ready flag to False to indicate that the connection is no longer ready. - Retrieves the WebSocket connection using get_connection. - Closes the WebSocket connection. - Triggers the closeConnectionError event if an error occurs during the closing process and raises the exception.

async def recover_connection(self): 
207    async def recover_connection(self):
208        try:
209            ws_connection_info = self.get_web_socket_token()
210            recovered_connection_info = await self.open_connection(
211                ws_connection_info["uri"], ws_connection_info["ws_access_token"]
212            )
213            # if recovered_connection_info['recoveryState'] === "Successful", then subscription is restored
214            # otherwise, need to create a new subscription
215            # IMPORTANT: WebSocket creation is successful if it doesn't raise any exception
216            return recovered_connection_info
217        except Exception as e:
218            self.trigger(WebSocketEvents.recoverConnectionError, e)
219            raise
async def send_message(self, message): 
221    async def send_message(self, message):
222        """
223            Sends a message over the WebSocket connection.
224
225            Args:
226                message (Any): The message to be sent.
227
228            Raises:
229                Exception: If any error occurs during the process or if the connection is not ready after multiple attempts.
230
231            Note:
232                - Checks if the WebSocket connection is ready (`_is_ready` flag).
233                - If the connection is ready, resets the send attempt counter and sends the message.
234                - If the connection is not ready, retries after a delay and increments the send attempt counter.
235                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
236        """
237        try:
238            if self._is_ready:
239                self._send_attempt_counter = 0
240                requestBodyJson = json.dumps(message)
241                await self.get_connection().send(requestBodyJson)
242            else:
243                await asyncio.sleep(1)
244                await self.send_message(message)
245                self._send_attempt_counter += 1
246                if(self._send_attempt_counter > 10):
247                    self.trigger(WebSocketEvents.connectionNotReady)
248                    self._send_attempt_counter = 0
249                    raise
250        except Exception as e:
251            self.trigger(WebSocketEvents.sendMessageError, e)
252            raise

Sends a message over the WebSocket connection.

Args: message (Any): The message to be sent.

Raises: Exception: If any error occurs during the process or if the connection is not ready after multiple attempts.

Note: - Checks if the WebSocket connection is ready (_is_ready flag). - If the connection is ready, resets the send attempt counter and sends the message. - If the connection is not ready, retries after a delay and increments the send attempt counter. - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.

async def create_subscription(self, events): 
254    async def create_subscription(self, events):
255        """
256            Creates a subscription to WebSocket events.
257
258            Args:
259                events (list): A required, non-empty list of events to subscribe to.
260
261            Raises:
262                Exception: If any error occurs during the process, if events are omitted, None, or empty, if a subscription creation is already in progress, or if a subscription already exists.
263
264            Note:
265                - Events are required: an omitted argument is rejected by ordinary Python argument checking, and None or empty events raise "Events are undefined" before any request is sent or any receive listener is attached.
266                - The client retains and reuses a single subscription object; a retry after a failed creation and a creation after removal reuse the retained object's existing registration flow with the newly supplied events.
267                - Rejects a call while a subscription creation is already in progress with "WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying".
268                - Rejects a call after a subscription has been created with "A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another".
269                - Rejections are delivered through the createSubscriptionError event and raised to the caller.
270                - If the WebSocket connection is ready (`_is_ready` flag), resets the send attempt counter and registers the retained subscription with the specified events.
271                - If the connection is not ready, retries after a delay and increments the send attempt counter.
272                - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.
273        """
274        try:
275            if not events or len(events) == 0:
276                raise Exception("Events are undefined")
277
278            if self._subscription is None:
279                self._subscription = WebSocketSubscription(self)
280            subscription = self._subscription
281
282            if subscription._pending_creation_message_id is not None:
283                raise Exception("WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying")
284            if subscription.get_subscription_info() is not None:
285                raise Exception("A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another")
286
287            if self._is_ready:
288                self._send_attempt_counter = 0
289                await subscription.register(events)
290            else:
291                await asyncio.sleep(1)
292                await self.create_subscription(events)
293                self._send_attempt_counter += 1
294                if(self._send_attempt_counter > 10):
295                    self.trigger(WebSocketEvents.connectionNotReady)
296                    self._send_attempt_counter = 0
297                    raise
298
299        except Exception as e:
300            self.trigger(WebSocketEvents.createSubscriptionError, e)
301            raise

Creates a subscription to WebSocket events.

Args: events (list): A required, non-empty list of events to subscribe to.

Raises: Exception: If any error occurs during the process, if events are omitted, None, or empty, if a subscription creation is already in progress, or if a subscription already exists.

Note: - Events are required: an omitted argument is rejected by ordinary Python argument checking, and None or empty events raise "Events are undefined" before any request is sent or any receive listener is attached. - The client retains and reuses a single subscription object; a retry after a failed creation and a creation after removal reuse the retained object's existing registration flow with the newly supplied events. - Rejects a call while a subscription creation is already in progress with "WebSocket subscription creation is already in progress; wait for subscriptionCreated or createSubscriptionError before retrying". - Rejects a call after a subscription has been created with "A WebSocket subscription already exists; use update_subscription() to change its events or remove_subscription() before creating another". - Rejections are delivered through the createSubscriptionError event and raised to the caller. - If the WebSocket connection is ready (_is_ready flag), resets the send attempt counter and registers the retained subscription with the specified events. - If the connection is not ready, retries after a delay and increments the send attempt counter. - If the send attempt counter exceeds a threshold, triggers the connectionNotReady event and raises an exception.

async def update_subscription(self, subscription, events=None): 
303    async def update_subscription(self, subscription, events=None):
304        """
305            Updates an existing WebSocket subscription with new events.
306
307            Args:
308                subscription : The WebSocket subscription to update.
309                events (list, optional): A list of events to update the subscription with. Default is None.
310
311            Returns:
312                WebSocketSubscription: The updated WebSocket subscription.
313
314            Raises:
315                Exception: If any error occurs during the process.
316
317            Note:
318                - Updates the specified WebSocket subscription with the new events provided.
319                - If the update is successful, returns the updated WebSocket subscription.
320                - If an error occurs during the update process, triggers the updateSubscriptionError event and raises an exception.
321        """
322        try:
323            await subscription.update(events)
324            return subscription
325        except Exception as e:
326            self.trigger(WebSocketEvents.updateSubscriptionError, e)
327            raise

Updates an existing WebSocket subscription with new events.

Args: subscription : The WebSocket subscription to update. events (list, optional): A list of events to update the subscription with. Default is None.

Returns: WebSocketSubscription: The updated WebSocket subscription.

Raises: Exception: If any error occurs during the process.

Note: - Updates the specified WebSocket subscription with the new events provided. - If the update is successful, returns the updated WebSocket subscription. - If an error occurs during the update process, triggers the updateSubscriptionError event and raises an exception.

async def remove_subscription(self, subscription): 
329    async def remove_subscription(self, subscription):
330        """
331            Removes an existing WebSocket subscription.
332
333            Args:
334                subscription : The WebSocket subscription to remove.
335
336            Raises:
337                Exception: If any error occurs during the removal process.
338
339            Note:
340                - Removes the specified WebSocket subscription.
341                - If the removal is successful, the subscription is effectively unsubscribed from the events it was subscribed to.
342                - If an error occurs during the removal process, triggers the removeSubscriptionError event and raises an exception.
343        """
344        try:
345            await subscription.remove()
346        except Exception as e:
347            self.trigger(WebSocketEvents.removeSubscriptionError, e)
348            raise

Removes an existing WebSocket subscription.

Args: subscription : The WebSocket subscription to remove.

Raises: Exception: If any error occurs during the removal process.

Note: - Removes the specified WebSocket subscription. - If the removal is successful, the subscription is effectively unsubscribed from the events it was subscribed to. - If an error occurs during the removal process, triggers the removeSubscriptionError event and raises an exception.

Inherited Members
observable.core.Observable
events
on
off
once