Ported more integration tests

- Fixed a bug where an asyncio Lock was tried to be acquired without a release
- Store asyncio tasks in order not to lose them
- Removed start/shutdown from AsyncioReactor

Author: yuce

hazelcast/asyncio/client.py

@@ -162,7 +162,6 @@ def _init_context(self):
         )
 
     async def _start(self):
-        self._reactor.start()
         try:
             self._internal_lifecycle_service.start()
             self._invocation_service.start()
@@ -250,7 +249,6 @@ async def shutdown(self) -> None:
                 await self._connection_manager.shutdown()
                 self._invocation_service.shutdown()
                 self._statistics.shutdown()
-                self._reactor.shutdown()
                 self._internal_lifecycle_service.fire_lifecycle_event(LifecycleState.SHUTDOWN)
 
     @property

hazelcast/internal/asyncio_connection.py

@@ -185,6 +185,9 @@ def __init__(
         self._use_public_ip = (
             isinstance(address_provider, DefaultAddressProvider) and config.use_public_ip
         )
+        # asyncio tasks are weakly referenced
+        # storing tasks here in order not to lose them midway
+        self._tasks = set()
 
     def add_listener(self, on_connection_opened=None, on_connection_closed=None):
         """Registers a ConnectionListener.
@@ -315,22 +318,21 @@ async def on_connection_close(self, closed_connection):
         disconnected = False
         removed = False
         trigger_reconnection = False
-        async with self._lock:
-            connection = self.active_connections.get(remote_uuid, None)
-            if connection == closed_connection:
-                self.active_connections.pop(remote_uuid, None)
-                removed = True
-                _logger.info(
-                    "Removed connection to %s:%s, connection: %s",
-                    remote_address,
-                    remote_uuid,
-                    connection,
-                )
+        connection = self.active_connections.get(remote_uuid, None)
+        if connection == closed_connection:
+            self.active_connections.pop(remote_uuid, None)
+            removed = True
+            _logger.info(
+                "Removed connection to %s:%s, connection: %s",
+                remote_address,
+                remote_uuid,
+                connection,
+            )
 
-                if not self.active_connections:
-                    trigger_reconnection = True
-                    if self._client_state == ClientState.INITIALIZED_ON_CLUSTER:
-                        disconnected = True
+            if not self.active_connections:
+                trigger_reconnection = True
+                if self._client_state == ClientState.INITIALIZED_ON_CLUSTER:
+                    disconnected = True
 
         if disconnected:
             self._lifecycle_service.fire_lifecycle_event(LifecycleState.DISCONNECTED)
@@ -813,6 +815,9 @@ def __init__(self, connection_manager, client, config, reactor, invocation_servi
         self._heartbeat_timeout = config.heartbeat_timeout
         self._heartbeat_interval = config.heartbeat_interval
         self._heartbeat_task: asyncio.Task | None = None
+        # asyncio tasks are weakly referenced
+        # storing tasks here in order not to lose them midway
+        self._tasks = set()
 
     def start(self):
         """Starts sending periodic HeartBeat operations."""
@@ -852,7 +857,9 @@ async def _check_connection(self, now, connection):
         if (now - connection.last_write_time) > self._heartbeat_interval:
             request = client_ping_codec.encode_request()
             invocation = Invocation(request, connection=connection, urgent=True)
-            asyncio.create_task(self._invocation_service.ainvoke(invocation))
+            task = asyncio.create_task(self._invocation_service.ainvoke(invocation))
+            self._tasks.add(task)
+            task.add_done_callback(self._tasks.discard)
 
 
 _frame_header = struct.Struct("<iH")

hazelcast/internal/asyncio_reactor.py

@@ -17,22 +17,13 @@
 
 class AsyncioReactor:
     def __init__(self, loop: AbstractEventLoop | None = None):
-        self._is_live = False
         self._loop = loop or asyncio.get_running_loop()
         self._bytes_sent = 0
         self._bytes_received = 0
 
     def add_timer(self, delay, callback):
         return self._loop.call_later(delay, callback)
 
-    def start(self):
-        self._is_live = True
-
-    def shutdown(self):
-        if not self._is_live:
-            return
-        # TODO: cancel tasks
-
     async def connection_factory(
         self, connection_manager, connection_id, address: Address, network_config, message_callback
     ):
@@ -174,6 +165,9 @@ def __init__(self, conn: AsyncioConnection):
         self._write_buf_size = 0
         self._recv_buf = None
         self._alive = True
+        # asyncio tasks are weakly referenced
+        # storing tasks here in order not to lose them midway
+        self._tasks: set = set()
 
     def connection_made(self, transport: transports.BaseTransport):
         self._transport = transport
@@ -184,7 +178,9 @@ def connection_made(self, transport: transports.BaseTransport):
 
     def connection_lost(self, exc):
         self._alive = False
-        self._conn._loop.create_task(self._conn.close_connection(str(exc), None))
+        task = self._conn._loop.create_task(self._conn.close_connection(str(exc), None))
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
         return False
 
     def close(self):