Fix frame reader

hazelcast/connection.py

@@ -693,7 +693,8 @@ def __init__(self, builder):
         self._builder = builder
         self._bytes_read = 0
         self._bytes_written = 0
-        self._frame_size = 0
+        # Size of the frame excluding the header (SIZE_OF_FRAME_LENGTH_AND_FLAGS bytes)
+        self._frame_size = -1
         self._frame_flags = 0
         self._message = None
 
@@ -719,22 +720,21 @@ def _read_message(self):
                 return None
 
     def _read_frame(self):
-        n = self.length
-        if n < SIZE_OF_FRAME_LENGTH_AND_FLAGS:
-            # we don't have even the frame length and flags ready
-            return False
+        if self._frame_size == -1:
+            if self.length < SIZE_OF_FRAME_LENGTH_AND_FLAGS:
+                # we don't have even the frame length and flags ready
+                return False
 
-        if self._frame_size == 0:
             self._read_frame_size_and_flags()
 
-        if n < self._frame_size:
+        if self.length < self._frame_size:
             return False
 
         self._buf.seek(self._bytes_read)
-        size = self._frame_size - SIZE_OF_FRAME_LENGTH_AND_FLAGS
+        size = self._frame_size
         data = self._buf.read(size)
         self._bytes_read += size
-        self._frame_size = 0
+        self._frame_size = -1
         # No need to reset flags since it will be overwritten on the next read_frame_size_and_flags call
         frame = Frame(data, self._frame_flags)
         if not self._message:
@@ -747,6 +747,7 @@ def _read_frame_size_and_flags(self):
         self._buf.seek(self._bytes_read)
         header_data = self._buf.read(SIZE_OF_FRAME_LENGTH_AND_FLAGS)
         self._frame_size, self._frame_flags = _frame_header.unpack_from(header_data, 0)
+        self._frame_size -= SIZE_OF_FRAME_LENGTH_AND_FLAGS
         self._bytes_read += SIZE_OF_FRAME_LENGTH_AND_FLAGS
 
     def _reset(self):

tests/integration/backward_compatible/proxy/map_test.py

@@ -11,7 +11,13 @@
     write_string_to_output,
     read_string_from_input,
 )
-from tests.util import random_string, event_collector, fill_map, is_server_version_older_than
+from tests.util import (
+    random_string,
+    event_collector,
+    fill_map,
+    is_server_version_older_than,
+    get_current_timestamp,
+)
 from hazelcast import six
 from hazelcast.six.moves import range
 
@@ -396,6 +402,13 @@ def test_put_get(self):
         self.assertIsNone(self.map.put("key", "value"))
         self.assertEqual(self.map.get("key"), "value")
 
+    def test_put_get_large_payload(self):
+        payload = bytearray(os.urandom(16 * 1024 * 1024))
+        start = get_current_timestamp()
+        self.assertIsNone(self.map.put("key", payload))
+        self.assertEqual(self.map.get("key"), payload)
+        self.assertLessEqual(get_current_timestamp() - start, 5)
+
     def test_put_get2(self):
         val = "x" * 5000