Add variable length JSON string support (#246)

Author: mp-e

pymysqlreplication/packet.py

@@ -252,6 +252,22 @@ def read_length_coded_pascal_string(self, size):
         length = self.read_uint_by_size(size)
         return self.read(length)
 
+    def read_variable_length_string(self):
+        """Read a variable length string where the first 1-5 bytes stores the
+        length of the string.
+        
+        For each byte, the first bit being high indicates another byte must be
+        read.
+        """
+        byte = 0x80
+        length = 0
+        bits_read = 0
+        while byte & 0x80 != 0:
+            byte = byte2int(self.read(1))
+            length = length | ((byte & 0x7f) << bits_read)
+            bits_read = bits_read + 7
+        return self.read(length)
+
     def read_int24(self):
         a, b, c = struct.unpack("BBB", self.read(3))
         res = a | (b << 8) | (c << 16)
@@ -342,7 +358,7 @@ def read_binary_json_type(self, t, length):
         elif t in (JSONB_TYPE_SMALL_ARRAY, JSONB_TYPE_LARGE_ARRAY):
             return self.read_binary_json_array(length - 1, large)
         elif t in (JSONB_TYPE_STRING,):
-            return self.read_length_coded_pascal_string(1)
+            return self.read_variable_length_string()
         elif t in (JSONB_TYPE_LITERAL,):
             value = self.read_uint8()
             if value == JSONB_LITERAL_NULL:

pymysqlreplication/tests/test_data_type.py

@@ -513,6 +513,16 @@ def test_json_unicode(self):
         event = self.create_and_insert_value(create_query, insert_query)
         self.assertEqual(event.rows[0]["values"]["value"][b"miam"], u'🍔'.encode('utf8'))
 
+    def test_json_long_string(self):
+        if not self.isMySQL57():
+            self.skipTest("Json is only supported in mysql 5.7")
+        create_query = "CREATE TABLE test (id int, value json);"
+        # The string length needs to be larger than what can fit in a single byte.
+        string_value = "super_long_string" * 100
+        insert_query = "INSERT INTO test (id, value) VALUES (1, '{\"my_key\": \"%s\"}');" % (string_value,)
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.rows[0]["values"]["value"], to_binary_dict({"my_key": string_value}))
+
     def test_null(self):
         create_query = "CREATE TABLE test ( \
             test TINYINT NULL DEFAULT NULL, \