fix: Do not supress callback exceptions (#332)

* feat: do not supress callback exceptions raised from calling callbacks

* fix: add explicit clauses for ConnectionClosedOK in send and heartbeat

* fix: change import order

* fix: apply ruff format

Author: o-santi

realtime/_async/client.py

@@ -10,6 +10,7 @@
 from websockets import connect
 from websockets.client import ClientProtocol
 
+from ..exceptions import NotConnectedError
 from ..message import Message
 from ..transformers import http_endpoint_url
 from ..types import (
@@ -92,7 +93,7 @@ async def _listen(self) -> None:
         """
 
         if not self._ws_connection:
-            raise Exception("WebSocket connection not established")
+            raise NotConnectedError("_listen")
 
         try:
             async for msg in self._ws_connection:
@@ -103,7 +104,7 @@ async def _listen(self) -> None:
 
                 if channel:
                     channel._trigger(msg.event, msg.payload, msg.ref)
-        except Exception as e:
+        except websockets.exceptions.ConnectionClosedError as e:
             await self._on_connect_error(e)
 
     async def _reconnect(self) -> None:
@@ -186,19 +187,18 @@ async def _on_connect(self) -> None:
         self._heartbeat_task = asyncio.create_task(self._heartbeat())
         await self._flush_send_buffer()
 
-    async def _on_connect_error(self, e: Exception) -> None:
-        if isinstance(e, websockets.exceptions.ConnectionClosedError):
-            logger.error(
-                f"WebSocket connection closed with code: {e.code}, reason: {e.reason}"
-            )
+    async def _on_connect_error(
+        self, e: websockets.exceptions.ConnectionClosedError
+    ) -> None:
+        logger.error(
+            f"WebSocket connection closed with code: {e.code}, reason: {e.reason}"
+        )
 
-            if self.auto_reconnect:
-                logger.info("Initiating auto-reconnect sequence...")
-                await self._reconnect()
-            else:
-                logger.error("Auto-reconnect disabled, terminating connection")
+        if self.auto_reconnect:
+            logger.info("Initiating auto-reconnect sequence...")
+            await self._reconnect()
         else:
-            logger.error(f"Error on connect: {e}")
+            logger.error("Auto-reconnect disabled, terminating connection")
 
     async def _flush_send_buffer(self):
         if self.is_connected and len(self.send_buffer) > 0:
@@ -232,7 +232,7 @@ async def close(self) -> None:
 
     async def _heartbeat(self) -> None:
         if not self._ws_connection:
-            raise Exception("WebSocket connection not established")
+            raise NotConnectedError("_heartbeat")
 
         while self.is_connected:
             try:
@@ -245,8 +245,10 @@ async def _heartbeat(self) -> None:
                 await self.send(data)
                 await asyncio.sleep(max(self.hb_interval, 15))
 
-            except Exception as e:
+            except websockets.exceptions.ConnectionClosedError as e:
                 await self._on_connect_error(e)
+            except websockets.exceptions.ConnectionClosedOK as e:
+                pass
 
     def channel(
         self, topic: str, params: Optional[RealtimeChannelOptions] = None
@@ -344,14 +346,14 @@ async def send(self, message: Dict[str, Any]) -> None:
 
         async def send_message():
             if not self._ws_connection:
-                raise Exception(
-                    "WebSocket connection not established, a connection is expected to be established before sending a message"
-                )
+                raise NotConnectedError("_send")
 
             try:
                 await self._ws_connection.send(message)
-            except Exception as e:
+            except websockets.exceptions.ConnectionClosedError as e:
                 await self._on_connect_error(e)
+            except websockets.exceptions.ConnectionClosedOK:
+                pass
 
         if self.is_connected:
             await send_message()

realtime/exceptions.py

@@ -7,7 +7,7 @@ def __init__(self, func_name: str):
         self.offending_func_name: str = func_name
 
     def __str__(self):
-        return f"A WS connection has not been established. Ensure you call RealtimeClient.connect() before calling RealtimeClient.{self.offending_func_name}()"
+        return f"A WS connection has not been established. Ensure you call AsyncRealtimeClient.connect() before calling AsyncRealtimeClient.{self.offending_func_name}()"
 
 
 class AuthorizationError(Exception):