Using TransportConnectionFailed instead of TransportClosed

Now the reconnecting session will reconnect as soon as it detects
that the connection failed.

Better log messages

Author: leszekhanusz

gql/client.py

@@ -35,7 +35,7 @@
 
 from .graphql_request import GraphQLRequest
 from .transport.async_transport import AsyncTransport
-from .transport.exceptions import TransportClosed, TransportQueryError
+from .transport.exceptions import TransportConnectionFailed, TransportQueryError
 from .transport.local_schema import LocalSchemaTransport
 from .transport.transport import Transport
 from .utilities import build_client_schema, get_introspection_query_ast
@@ -1730,6 +1730,7 @@ async def _connection_loop(self):
             # Then wait for the reconnect event
             self._reconnect_request_event.clear()
             await self._reconnect_request_event.wait()
+            await self.transport.close()
 
     async def start_connecting_task(self):
         """Start the task responsible to restart the connection
@@ -1758,7 +1759,7 @@ async def _execute_once(
         **kwargs: Any,
     ) -> ExecutionResult:
         """Same Coroutine as parent method _execute but requesting a
-        reconnection if we receive a TransportClosed exception.
+        reconnection if we receive a TransportConnectionFailed exception.
         """
 
         try:
@@ -1770,7 +1771,7 @@ async def _execute_once(
                 parse_result=parse_result,
                 **kwargs,
             )
-        except TransportClosed:
+        except TransportConnectionFailed:
             self._reconnect_request_event.set()
             raise
 
@@ -1786,7 +1787,8 @@ async def _execute(
         **kwargs: Any,
     ) -> ExecutionResult:
         """Same Coroutine as parent, but with optional retries
-        and requesting a reconnection if we receive a TransportClosed exception.
+        and requesting a reconnection if we receive a
+        TransportConnectionFailed exception.
         """
 
         return await self._execute_with_retries(
@@ -1808,7 +1810,7 @@ async def _subscribe(
         **kwargs: Any,
     ) -> AsyncGenerator[ExecutionResult, None]:
         """Same Async generator as parent method _subscribe but requesting a
-        reconnection if we receive a TransportClosed exception.
+        reconnection if we receive a TransportConnectionFailed exception.
         """
 
         inner_generator: AsyncGenerator[ExecutionResult, None] = super()._subscribe(
@@ -1824,7 +1826,7 @@ async def _subscribe(
             async for result in inner_generator:
                 yield result
 
-        except TransportClosed:
+        except TransportConnectionFailed:
             self._reconnect_request_event.set()
             raise
 

gql/transport/common/adapters/aiohttp.py

@@ -178,12 +178,14 @@ async def send(self, message: str) -> None:
             TransportConnectionFailed: If connection closed
         """
         if self.websocket is None:
-            raise TransportConnectionFailed("Connection is already closed")
+            raise TransportConnectionFailed("WebSocket connection is already closed")
 
         try:
             await self.websocket.send_str(message)
-        except ConnectionResetError as e:
-            raise TransportConnectionFailed("Connection was closed") from e
+        except Exception as e:
+            raise TransportConnectionFailed(
+                f"Error trying to send data: {type(e).__name__}"
+            ) from e
 
     async def receive(self) -> str:
         """Receive message from the WebSocket server.
@@ -200,6 +202,9 @@ async def receive(self) -> str:
             raise TransportConnectionFailed("Connection is already closed")
 
         while True:
+            # Should not raise any exception:
+            # https://docs.aiohttp.org/en/stable/_modules/aiohttp/client_ws.html
+            #                                           #ClientWebSocketResponse.receive
             ws_message = await self.websocket.receive()
 
             # Ignore low-level ping and pong received

gql/transport/common/adapters/websockets.py

@@ -86,12 +86,14 @@ async def send(self, message: str) -> None:
             TransportConnectionFailed: If connection closed
         """
         if self.websocket is None:
-            raise TransportConnectionFailed("Connection is already closed")
+            raise TransportConnectionFailed("WebSocket connection is already closed")
 
         try:
             await self.websocket.send(message)
         except Exception as e:
-            raise TransportConnectionFailed("Connection was closed") from e
+            raise TransportConnectionFailed(
+                f"Error trying to send data: {type(e).__name__}"
+            ) from e
 
     async def receive(self) -> str:
         """Receive message from the WebSocket server.
@@ -111,7 +113,9 @@ async def receive(self) -> str:
         try:
             data = await self.websocket.recv()
         except Exception as e:
-            raise TransportConnectionFailed("Connection was closed") from e
+            raise TransportConnectionFailed(
+                f"Error trying to receive data: {type(e).__name__}"
+            ) from e
 
         # websocket.recv() can return either str or bytes
         # In our case, we should receive only str here

gql/transport/common/base.py

@@ -127,11 +127,13 @@ async def _send(self, message: str) -> None:
         """Send the provided message to the adapter connection and log the message"""
 
         if not self._connected:
-            raise TransportClosed(
-                "Transport is not connected"
-            ) from self.close_exception
+            if isinstance(self.close_exception, TransportConnectionFailed):
+                raise self.close_exception
+            else:
+                raise TransportConnectionFailed() from self.close_exception
 
         try:
+            # Can raise TransportConnectionFailed
             await self.adapter.send(message)
             log.info(">>> %s", message)
         except TransportConnectionFailed as e:
@@ -143,7 +145,7 @@ async def _receive(self) -> str:
 
         # It is possible that the connection has been already closed in another task
         if not self._connected:
-            raise TransportClosed("Transport is already closed")
+            raise TransportConnectionFailed() from self.close_exception
 
         # Wait for the next frame.
         # Can raise TransportConnectionFailed or TransportProtocolError
@@ -214,8 +216,6 @@ async def _receive_data_loop(self) -> None:
                 except (TransportConnectionFailed, TransportProtocolError) as e:
                     await self._fail(e, clean_close=False)
                     break
-                except TransportClosed:
-                    break
 
                 # Parse the answer
                 try:
@@ -503,9 +503,10 @@ async def _close_coro(self, e: Exception, clean_close: bool = True) -> None:
                 except Exception as exc:  # pragma: no cover
                     log.warning("Ignoring exception in _clean_close: " + repr(exc))
 
-            log.debug(
-                f"_close_coro: sending exception to {len(self.listeners)} listeners"
-            )
+            if log.isEnabledFor(logging.DEBUG):
+                log.debug(
+                    f"_close_coro: sending exception to {len(self.listeners)} listeners"
+                )
 
             # Send an exception to all remaining listeners
             for query_id, listener in self.listeners.items():
@@ -532,7 +533,15 @@ async def _close_coro(self, e: Exception, clean_close: bool = True) -> None:
         log.debug("_close_coro: exiting")
 
     async def _fail(self, e: Exception, clean_close: bool = True) -> None:
-        log.debug("_fail: starting with exception: " + repr(e))
+        if log.isEnabledFor(logging.DEBUG):
+            import inspect
+
+            current_frame = inspect.currentframe()
+            assert current_frame is not None
+            caller_frame = current_frame.f_back
+            assert caller_frame is not None
+            caller_name = inspect.getframeinfo(caller_frame).function
+            log.debug(f"_fail from {caller_name}: " + repr(e))
 
         if self.close_task is None:
 

tests/test_aiohttp_websocket_exceptions.py

@@ -301,6 +301,15 @@ async def test_aiohttp_websocket_server_closing_after_ack(aiohttp_client_and_ser
 
     query = gql("query { hello }")
 
+    print("\n Trying to execute first query.\n")
+
+    with pytest.raises(TransportConnectionFailed):
+        await session.execute(query)
+
+    await session.transport.wait_closed()
+
+    print("\n Trying to execute second query.\n")
+
     with pytest.raises(TransportConnectionFailed):
         await session.execute(query)
 

tests/test_aiohttp_websocket_graphqlws_exceptions.py

@@ -5,7 +5,6 @@
 
 from gql import Client, gql
 from gql.transport.exceptions import (
-    TransportClosed,
     TransportConnectionFailed,
     TransportProtocolError,
     TransportQueryError,
@@ -264,10 +263,14 @@ async def test_aiohttp_websocket_graphqlws_server_closing_after_ack(
 
     query = gql("query { hello }")
 
+    print("\n Trying to execute first query.\n")
+
     with pytest.raises(TransportConnectionFailed):
         await session.execute(query)
 
     await session.transport.wait_closed()
 
-    with pytest.raises(TransportClosed):
+    print("\n Trying to execute second query.\n")
+
+    with pytest.raises(TransportConnectionFailed):
         await session.execute(query)

tests/test_aiohttp_websocket_graphqlws_subscription.py

@@ -11,7 +11,7 @@
 from gql.client import AsyncClientSession
 from gql.transport.exceptions import TransportConnectionFailed, TransportServerError
 
-from .conftest import MS, PyPy, WebSocketServerHelper
+from .conftest import MS, WebSocketServerHelper
 
 # Marking all tests in this file with the aiohttp AND websockets marker
 pytestmark = [pytest.mark.aiohttp, pytest.mark.websockets]
@@ -821,7 +821,6 @@ async def test_aiohttp_websocket_graphqlws_subscription_reconnecting_session(
 ):
 
     from gql.transport.aiohttp_websockets import AIOHTTPWebsocketsTransport
-    from gql.transport.exceptions import TransportClosed
 
     path = "/graphql"
     url = f"ws://{graphqlws_server.hostname}:{graphqlws_server.port}{path}"
@@ -839,56 +838,64 @@ async def test_aiohttp_websocket_graphqlws_subscription_reconnecting_session(
         reconnecting=True, retry_connect=False, retry_execute=False
     )
 
-    # First we make a subscription which will cause a disconnect in the backend
-    # (count=8)
-    try:
-        print("\nSUBSCRIPTION_1_WITH_DISCONNECT\n")
-        async for result in session.subscribe(subscription_with_disconnect):
-            pass
-    except TransportConnectionFailed:
-        pass
-
-    await asyncio.sleep(50 * MS)
-
-    # Then with the same session handle, we make a subscription or an execute
-    # which will detect that the transport is closed so that the client could
-    # try to reconnect
-    generator = None
+    # First we make a query or subscription which will cause a disconnect
+    # in the backend (count=8)
     try:
         if execute_instead_of_subscribe:
-            print("\nEXECUTION_2\n")
-            await session.execute(subscription)
+            print("\nEXECUTION_1\n")
+            await session.execute(subscription_with_disconnect)
         else:
-            print("\nSUBSCRIPTION_2\n")
-            generator = session.subscribe(subscription)
-            async for result in generator:
+            print("\nSUBSCRIPTION_1_WITH_DISCONNECT\n")
+            async for result in session.subscribe(subscription_with_disconnect):
                 pass
-    except (TransportClosed, TransportConnectionFailed):
-        if generator:
-            await generator.aclose()
+    except TransportConnectionFailed:
         pass
 
-    timeout = 50
-
-    if PyPy:
-        timeout = 500
+    # Wait for disconnect
+    for i in range(200):
+        await asyncio.sleep(1 * MS)
+        if not transport._connected:
+            print(f"\nDisconnected in {i+1} MS")
+            break
 
-    await asyncio.sleep(timeout * MS)
+    assert transport._connected is False
 
-    # And finally with the same session handle, we make a subscription
-    # which works correctly
-    print("\nSUBSCRIPTION_3\n")
-    generator = session.subscribe(subscription)
-    async for result in generator:
+    # Wait for reconnect
+    for i in range(200):
+        await asyncio.sleep(1 * MS)
+        if transport._connected:
+            print(f"\nConnected again in {i+1} MS")
+            break
 
-        number = result["number"]
-        print(f"Number received: {number}")
+    assert transport._connected is True
+
+    # Then after the reconnection, we make a query or a subscription
+    if execute_instead_of_subscribe:
+        print("\nEXECUTION_2\n")
+        result = await session.execute(subscription)
+        assert result["number"] == 10
+    else:
+        print("\nSUBSCRIPTION_2\n")
+        generator = session.subscribe(subscription)
+        async for result in generator:
+            number = result["number"]
+            print(f"Number received: {number}")
 
-        assert number == count
-        count -= 1
+            assert number == count
+            count -= 1
 
-    await generator.aclose()
+        await generator.aclose()
 
-    assert count == -1
+        assert count == -1
 
+    # Close the reconnecting session
     await client.close_async()
+
+    # Wait for disconnect
+    for i in range(200):
+        await asyncio.sleep(1 * MS)
+        if not transport._connected:
+            print(f"\nDisconnected in {i+1} MS")
+            break
+
+    assert transport._connected is False

tests/test_aiohttp_websocket_query.py

@@ -8,7 +8,6 @@
 from gql import Client, gql
 from gql.transport.exceptions import (
     TransportAlreadyConnected,
-    TransportClosed,
     TransportConnectionFailed,
     TransportQueryError,
     TransportServerError,
@@ -323,7 +322,7 @@ async def test_aiohttp_websocket_server_closing_after_first_query(
 
     # Now the server is closed but we don't know it yet, we have to send a query
     # to notice it and to receive the exception
-    with pytest.raises(TransportClosed):
+    with pytest.raises(TransportConnectionFailed):
         await session.execute(query)