Asyncio Module Cloud Support (#754)

Improved the stability of the asyncio client in this PR.

* Added support for non-blocking Hazelcast Cloud address resolution
* Ported `connection_manager_test.py` to asyncio
* Refactored `AsyncioConnection` to be more reliable by porting more
logic from `AsyncoreConnection`.
* Run all VectorCollection tests for 5.6.0

Author: yuce

Makefile

@@ -10,5 +10,8 @@ test:
 test-enterprise:
 	pytest --verbose
 
+test-asyncio:
+	pytest --verbose -k asyncio
+
 test-cover:
 	pytest --cov=hazelcast --cov-report=xml

examples/asyncio/cloud-discovery/hazelcast_cloud_discovery_example.py

@@ -0,0 +1,28 @@
+import asyncio
+
+from hazelcast.asyncio import HazelcastClient
+
+
+async def amain():
+    client = await HazelcastClient.create_and_start(
+        # Set up cluster name for authentication
+        cluster_name="asyncio",
+        # Set the token of your cloud cluster
+        cloud_discovery_token="wE1w1USF6zOnaLVjLZwbZHxEoZJhw43yyViTbe6UBTvz4tZniA",
+        ssl_enabled=True,
+        ssl_cafile="/path/to/ca.pem",
+        ssl_certfile="/path/to/cert.pem",
+        ssl_keyfile="/path/to/key.pem",
+        ssl_password="05dd4498c3f",
+    )
+    my_map = await client.get_map("map-on-the-cloud")
+    await my_map.put("key", "value")
+
+    value = await my_map.get("key")
+    print(value)
+
+    await client.shutdown()
+
+
+if __name__ == "__main__":
+    asyncio.run(amain())

examples/cloud-discovery/hazelcast_cloud_discovery_example.py

@@ -5,8 +5,6 @@
     cluster_name="YOUR_CLUSTER_NAME",
     # Set the token of your cloud cluster
     cloud_discovery_token="YOUR_CLUSTER_DISCOVERY_TOKEN",
-    # If you have enabled encryption for your cluster, also configure TLS/SSL for the client.
-    # Otherwise, skip options below.
     ssl_enabled=True,
     ssl_cafile="/path/to/ca.pem",
     ssl_certfile="/path/to/cert.pem",

hazelcast/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "6.0.0"
+__version__ = "5.6.0"
 
 # Set the default handler to "hazelcast" loggers
 # to avoid "No handlers could be found" warnings.

hazelcast/asyncio/client.py

@@ -6,7 +6,7 @@
 from hazelcast.internal.asyncio_cluster import ClusterService, _InternalClusterService
 from hazelcast.internal.asyncio_compact import CompactSchemaService
 from hazelcast.config import Config, IndexConfig
-from hazelcast.internal.asyncio_connection import ConnectionManager, DefaultAddressProvider
+from hazelcast.internal.asyncio_connection import ConnectionManager, DefaultAsyncioAddressProvider
 from hazelcast.core import DistributedObjectEvent, DistributedObjectInfo
 from hazelcast.discovery import HazelcastCloudAddressProvider
 from hazelcast.errors import IllegalStateError, InvalidConfigurationError
@@ -313,7 +313,7 @@ def _create_address_provider(self):
             connection_timeout = self._get_connection_timeout(config)
             return HazelcastCloudAddressProvider(cloud_discovery_token, connection_timeout)
 
-        return DefaultAddressProvider(cluster_members)
+        return DefaultAsyncioAddressProvider(cluster_members)
 
     def _create_client_name(self, client_id):
         client_name = self._config.client_name

hazelcast/internal/asyncio_connection.py

@@ -183,7 +183,7 @@ def __init__(
         self._cluster_id = None
         self._load_balancer = None
         self._use_public_ip = (
-            isinstance(address_provider, DefaultAddressProvider) and config.use_public_ip
+            isinstance(address_provider, DefaultAsyncioAddressProvider) and config.use_public_ip
         )
         # asyncio tasks are weakly referenced
         # storing tasks here in order not to lose them midway
@@ -385,9 +385,10 @@ async def _get_or_connect_to_address(self, address):
         for connection in list(self.active_connections.values()):
             if connection.remote_address == address:
                 return connection
-        translated = self._translate(address)
-        connection = await self._create_connection(translated)
-        response = await self._authenticate(connection)
+        translated = await self._translate(address)
+        connection = self._create_connection(translated)
+        await connection._create_task
+        response = self._authenticate(connection)
         await self._on_auth(response, connection)
         return connection
 
@@ -396,23 +397,24 @@ async def _get_or_connect_to_member(self, member):
         if connection:
             return connection
 
-        translated = self._translate_member_address(member)
-        connection = await self._create_connection(translated)
-        response = await self._authenticate(connection)
+        translated = await self._translate_member_address(member)
+        connection = self._create_connection(translated)
+        await connection._create_task
+        response = self._authenticate(connection)
         await self._on_auth(response, connection)
         return connection
 
-    async def _create_connection(self, address):
-        return await self._reactor.connection_factory(
+    def _create_connection(self, address):
+        return self._reactor.connection_factory(
             self,
             self._connection_id_generator.get_and_increment(),
             address,
             self._config,
             self._invocation_service.handle_client_message,
         )
 
-    def _translate(self, address):
-        translated = self._address_provider.translate(address)
+    async def _translate(self, address):
+        translated = await self._address_provider.translate(address)
         if not translated:
             raise ValueError(
                 "Address provider %s could not translate address %s"
@@ -421,15 +423,15 @@ def _translate(self, address):
 
         return translated
 
-    def _translate_member_address(self, member):
+    async def _translate_member_address(self, member):
         if self._use_public_ip:
             public_address = member.address_map.get(_CLIENT_PUBLIC_ENDPOINT_QUALIFIER, None)
             if public_address:
                 return public_address
 
             return member.address
 
-        return self._translate(member.address)
+        return await self._translate(member.address)
 
     async def _trigger_cluster_reconnection(self):
         if self._reconnect_mode == ReconnectMode.OFF:
@@ -529,7 +531,8 @@ async def _sync_connect_to_cluster(self):
                     if connection:
                         return
 
-                for address in self._get_possible_addresses():
+                addresses = await self._get_possible_addresses()
+                for address in addresses:
                     self._check_client_active()
                     if address in tried_addresses_per_attempt:
                         # We already tried this address on from the member list
@@ -614,6 +617,7 @@ def _authenticate(self, connection) -> asyncio.Future:
 
     async def _on_auth(self, response, connection):
         try:
+            response = await response
             response = client_authentication_codec.decode_response(response)
         except Exception as e:
             await connection.close_connection("Failed to authenticate connection", e)
@@ -790,8 +794,8 @@ def _check_client_active(self):
         if not self._lifecycle_service.running:
             raise HazelcastClientNotActiveError()
 
-    def _get_possible_addresses(self):
-        primaries, secondaries = self._address_provider.load_addresses()
+    async def _get_possible_addresses(self):
+        primaries, secondaries = await self._address_provider.load_addresses()
         if self._shuffle_member_list:
             # The relative order between primary and secondary addresses should
             # not be changed. So we shuffle the lists separately and then add
@@ -1028,17 +1032,13 @@ def __hash__(self):
         return self._id
 
 
-class DefaultAddressProvider:
-    """Provides initial addresses for client to find and connect to a node.
-
-    It also provides a no-op translator.
-    """
-
+class DefaultAsyncioAddressProvider:
     def __init__(self, addresses):
         self._addresses = addresses
 
-    def load_addresses(self):
+    async def load_addresses(self):
         """Returns the possible primary and secondary member addresses to connect to."""
+        # NOTE: This method is marked with async since the caller assumes that.
         configured_addresses = self._addresses
 
         if not configured_addresses:
@@ -1053,9 +1053,10 @@ def load_addresses(self):
 
         return primaries, secondaries
 
-    def translate(self, address):
+    async def translate(self, address):
         """No-op address translator.
 
         It is there to provide the same API with other address providers.
         """
+        # NOTE: This method is marked with async since the caller assumes that.
         return address

hazelcast/internal/asyncio_discovery.py

@@ -0,0 +1,58 @@
+import asyncio
+import logging
+
+from hazelcast.discovery import HazelcastCloudDiscovery
+
+_logger = logging.getLogger(__name__)
+
+
+class HazelcastCloudAddressProvider:
+    """Provides initial addresses for client to find and connect to a node
+    and resolves private IP addresses of Hazelcast Cloud service.
+    """
+
+    def __init__(self, token, connection_timeout):
+        self.cloud_discovery = HazelcastCloudDiscovery(token, connection_timeout)
+        self._private_to_public = dict()
+
+    async def load_addresses(self):
+        """Loads member addresses from Hazelcast Cloud endpoint.
+
+        Returns:
+            tuple[list[hazelcast.core.Address], list[hazelcast.core.Address]]: The possible member addresses
+                as primary addresses to connect to.
+        """
+        try:
+            nodes = await asyncio.to_thread(self.cloud_discovery.discover_nodes)
+            # Every private address is primary
+            return list(nodes.keys()), []
+        except Exception as e:
+            _logger.warning("Failed to load addresses from Hazelcast Cloud: %s", e)
+        return [], []
+
+    async def translate(self, address):
+        """Translates the given address to another address specific to network or service.
+
+        Args:
+            address (hazelcast.core.Address): Private address to be translated
+
+        Returns:
+            hazelcast.core.Address: New address if given address is known, otherwise returns None
+        """
+        if address is None:
+            return None
+
+        public_address = self._private_to_public.get(address, None)
+        if public_address:
+            return public_address
+
+        await self.refresh()
+
+        return self._private_to_public.get(address, None)
+
+    async def refresh(self):
+        """Refreshes the internal lookup table if necessary."""
+        try:
+            self._private_to_public = self.cloud_discovery.discover_nodes()
+        except Exception as e:
+            _logger.warning("Failed to load addresses from Hazelcast Cloud: %s", e)