Eagerly deserialize responses consisting of list of data [API-1964]

docs/serialization.rst

@@ -278,33 +278,6 @@ All of these APIs will work with the Compact serialization format, once it is
 promoted to the stable status.
 
 - Reading OBJECT columns of the SQL results
-- Listening for :class:`hazelcast.proxy.reliable_topic.ReliableTopic` messages
-- :func:`hazelcast.proxy.list.List.iterator`
-- :func:`hazelcast.proxy.list.List.list_iterator`
-- :func:`hazelcast.proxy.list.List.get_all`
-- :func:`hazelcast.proxy.list.List.sub_list`
-- :func:`hazelcast.proxy.map.Map.values`
-- :func:`hazelcast.proxy.map.Map.entry_set`
-- :func:`hazelcast.proxy.map.Map.execute_on_keys`
-- :func:`hazelcast.proxy.map.Map.key_set`
-- :func:`hazelcast.proxy.map.Map.project`
-- :func:`hazelcast.proxy.map.Map.execute_on_entries`
-- :func:`hazelcast.proxy.map.Map.get_all`
-- :func:`hazelcast.proxy.multi_map.MultiMap.remove_all`
-- :func:`hazelcast.proxy.multi_map.MultiMap.key_set`
-- :func:`hazelcast.proxy.multi_map.MultiMap.values`
-- :func:`hazelcast.proxy.multi_map.MultiMap.entry_set`
-- :func:`hazelcast.proxy.multi_map.MultiMap.get`
-- :func:`hazelcast.proxy.queue.Queue.iterator`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.values`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.entry_set`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.key_set`
-- :func:`hazelcast.proxy.set.Set.get_all`
-- :func:`hazelcast.proxy.ringbuffer.Ringbuffer.read_many`
-- :func:`hazelcast.proxy.transactional_map.TransactionalMap.values`
-- :func:`hazelcast.proxy.transactional_map.TransactionalMap.key_set`
-- :func:`hazelcast.proxy.transactional_multi_map.TransactionalMultiMap.get`
-- :func:`hazelcast.proxy.transactional_multi_map.TransactionalMultiMap.remove_all`
 
 IdentifiedDataSerializable Serialization
 ----------------------------------------

hazelcast/proxy/list.py

@@ -29,7 +29,7 @@
 from hazelcast.proxy.base import PartitionSpecificProxy, ItemEvent, ItemEventType
 from hazelcast.types import ItemType
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, ImmutableLazyDataList
+from hazelcast.util import check_not_none, deserialize_list_in_place
 
 
 class List(PartitionSpecificProxy["BlockingList"], typing.Generic[ItemType]):
@@ -247,9 +247,8 @@ def get_all(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_get_all_codec.decode_response(message), self._to_object
-            )
+            data_list = list_get_all_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_get_all_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -263,9 +262,8 @@ def iterator(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = list_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_iterator_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -337,9 +335,8 @@ def list_iterator(self, index: int = 0) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_list_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = list_list_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_list_iterator_codec.encode_request(self.name, index)
         return self._invoke(request, handler)
@@ -494,7 +491,8 @@ def sub_list(self, from_index: int, to_index: int) -> Future[typing.List[ItemTyp
         """
 
         def handler(message):
-            return ImmutableLazyDataList(list_sub_codec.decode_response(message), self._to_object)
+            data_list = list_sub_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_sub_codec.encode_request(self.name, from_index, to_index)
         return self._invoke(request, handler)

hazelcast/proxy/map.py

@@ -87,8 +87,9 @@
     check_not_none,
     thread_id,
     to_millis,
-    ImmutableLazyDataList,
     IterationType,
+    deserialize_entry_list_in_place,
+    deserialize_list_in_place,
 )
 
 
@@ -563,7 +564,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    entry_data_list = response["response"]
+                    return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
                 request = map_entries_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -573,17 +575,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.entry_set, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_entries_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    entry_data_list = map_entries_with_predicate_codec.decode_response(message)
+                    return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
                 request = map_entries_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_entry_set_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_entry_set_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_entry_set_codec.encode_request(self.name)
 
@@ -648,9 +648,8 @@ def execute_on_entries(
                 )
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_execute_with_predicate_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_execute_with_predicate_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_execute_with_predicate_codec.encode_request(
                 self.name, entry_processor_data, predicate_data
@@ -664,9 +663,8 @@ def handler(message):
                 )
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_execute_on_all_keys_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_execute_on_all_keys_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_execute_on_all_keys_codec.encode_request(self.name, entry_processor_data)
 
@@ -730,9 +728,8 @@ def execute_on_keys(
             return self._send_schema_and_retry(e, self.execute_on_keys, keys, entry_processor)
 
         def handler(message):
-            return ImmutableLazyDataList(
-                map_execute_on_keys_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = map_execute_on_keys_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         request = map_execute_on_keys_codec.encode_request(
             self.name, entry_processor_data, key_list
@@ -943,7 +940,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    data_list = response["response"]
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_key_set_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -953,17 +951,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.key_set, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_key_set_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    data_list = map_key_set_with_predicate_codec.decode_response(message)
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_key_set_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_key_set_codec.decode_response(message), self._to_object
-                )
+                data_list = map_key_set_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_key_set_codec.encode_request(self.name)
 
@@ -1067,9 +1063,8 @@ def project(
                 return self._send_schema_and_retry(e, self.project, projection, predicate)
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_project_with_predicate_codec.decode_response(message), self._to_object
-                )
+                data_list = map_project_with_predicate_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_project_with_predicate_codec.encode_request(
                 self.name, projection_data, predicate_data
@@ -1081,9 +1076,8 @@ def handler(message):
                 return self._send_schema_and_retry(e, self.project, projection, predicate)
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_project_codec.decode_response(message), self._to_object
-                )
+                data_list = map_project_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_project_codec.encode_request(self.name, projection_data)
 
@@ -1652,7 +1646,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    data_list = response["response"]
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_values_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -1662,17 +1657,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.values, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_values_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    data_list = map_values_with_predicate_codec.decode_response(message)
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_values_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_values_codec.decode_response(message), self._to_object
-                )
+                data_list = map_values_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_values_codec.encode_request(self.name)
 
@@ -1698,9 +1691,8 @@ def _get_all_internal(self, partition_to_keys, futures=None):
             futures = []
 
         def handler(message):
-            return ImmutableLazyDataList(
-                map_get_all_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = map_get_all_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         for partition_id, key_dict in partition_to_keys.items():
             request = map_get_all_codec.encode_request(self.name, key_dict.values())

hazelcast/proxy/multi_map.py

@@ -31,8 +31,13 @@
 from hazelcast.types import ValueType, KeyType
 from hazelcast.serialization.data import Data
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, thread_id, to_millis, ImmutableLazyDataList
-
+from hazelcast.util import (
+    check_not_none,
+    thread_id,
+    to_millis,
+    deserialize_list_in_place,
+    deserialize_entry_list_in_place,
+)
 
 EntryEventCallable = typing.Callable[[EntryEvent[KeyType, ValueType]], None]
 
@@ -213,9 +218,8 @@ def entry_set(self) -> Future[typing.List[typing.Tuple[KeyType, ValueType]]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_entry_set_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = multi_map_entry_set_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         request = multi_map_entry_set_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -246,9 +250,8 @@ def get(self, key: KeyType) -> Future[typing.Optional[typing.List[ValueType]]]:
             return self._send_schema_and_retry(e, self.get, key)
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_get_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_get_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_get_codec.encode_request(self.name, key_data, thread_id())
         return self._invoke_on_key(request, key_data, handler)
@@ -314,9 +317,8 @@ def key_set(self) -> Future[typing.List[KeyType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_key_set_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_key_set_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_key_set_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -410,9 +412,8 @@ def remove_all(self, key: KeyType) -> Future[typing.List[ValueType]]:
         check_not_none(key, "key can't be None")
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_remove_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_remove_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         try:
             key_data = self._to_data(key)
@@ -551,9 +552,8 @@ def values(self) -> Future[typing.List[ValueType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_values_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_values_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_values_codec.encode_request(self.name)
         return self._invoke(request, handler)

hazelcast/proxy/queue.py

@@ -26,7 +26,7 @@
 from hazelcast.proxy.base import PartitionSpecificProxy, ItemEvent, ItemEventType
 from hazelcast.types import ItemType
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, to_millis, ImmutableLazyDataList
+from hazelcast.util import check_not_none, to_millis, deserialize_list_in_place
 
 
 class Queue(PartitionSpecificProxy["BlockingQueue"], typing.Generic[ItemType]):
@@ -202,9 +202,8 @@ def iterator(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                queue_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = queue_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = queue_iterator_codec.encode_request(self.name)
         return self._invoke(request, handler)

hazelcast/proxy/reliable_topic.py

@@ -226,7 +226,7 @@ def _handle_next_batch(self, future):
             result = future.result()
 
             # Check if there are any messages lost since the last read
-            # and whether or not the listener can tolerate that.
+            # and whether the listener can tolerate that.
             lost_count = (result.next_sequence_to_read_from - result.read_count) - self._sequence
             if lost_count != 0 and not self._is_loss_tolerable(lost_count):
                 self.cancel()
@@ -254,7 +254,7 @@ def _handle_next_batch(self, future):
 
                     topic_message = TopicMessage(
                         self._topic_name,
-                        self._to_object(message.payload),
+                        message.payload,
                         message.publish_time,
                         member,
                     )