Merge branch 'master' into asyncio-module-integration-tests2

# Conflicts:
#	hazelcast/internal/asyncio_connection.py
#	hazelcast/internal/asyncio_reactor.py

Author: yuce

hazelcast/internal/asyncio_connection.py

@@ -187,6 +187,7 @@ def __init__(
         )
         # asyncio tasks are weakly referenced
         # storing tasks here in order not to lose them midway
+        # see: https: // docs.python.org / 3 / library / asyncio - task.html  # creating-tasks
         self._tasks = set()
 
     def add_listener(self, on_connection_opened=None, on_connection_closed=None):
@@ -318,21 +319,22 @@ async def on_connection_close(self, closed_connection):
         disconnected = False
         removed = False
         trigger_reconnection = False
-        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,
-            )
+        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,
+                )
 
-            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,7 @@ def __init__(self, connection_manager, client, config, reactor, invocation_servi
         self._heartbeat_task: asyncio.Task | None = None
         # asyncio tasks are weakly referenced
         # storing tasks here in order not to lose them midway
+        # see: https: // docs.python.org / 3 / library / asyncio - task.html  # creating-tasks
         self._tasks = set()
 
     def start(self):

hazelcast/internal/asyncio_reactor.py

@@ -175,6 +175,7 @@ def __init__(self, conn: AsyncioConnection):
         self._alive = True
         # asyncio tasks are weakly referenced
         # storing tasks here in order not to lose them midway
+        # see: https: // docs.python.org / 3 / library / asyncio - task.html  # creating-tasks
         self._tasks: set = set()
 
     def connection_made(self, transport: transports.BaseTransport):

tests/integration/asyncio/proxy/map_test.py

@@ -629,7 +629,7 @@ async def test_add_entry_listener_item_loaded(self):
         collector = event_collector()
         await self.map.add_entry_listener(include_value=True, loaded_func=collector)
         await self.map.put("key", "value", ttl=0.1)
-        time.sleep(2)
+        await asyncio.sleep(2)
         await self.map.get("key")
 
         def assert_event():