Add properties in gtid event

pymysqlreplication/event.py

@@ -108,6 +108,43 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
             self.last_committed = struct.unpack("<Q", self.packet.read(8))[0]
             self.sequence_number = struct.unpack("<Q", self.packet.read(8))[0]
 
+        format_string = "<Q"
+        read_immediate_commit_timestamp = self.packet.read(7)
+        self.immediate_commit_timestamp = read_immediate_commit_timestamp.ljust(
+            struct.calcsize(format_string), b"\x00"
+        )
+        self.immediate_commit_timestamp = struct.unpack(
+            format_string, self.immediate_commit_timestamp
+        )[0]
+        self.immediate_commit_timestamp = self.immediate_commit_timestamp & (
+            (1 << (8 * struct.calcsize(format_string))) - 1
+        )
+        self.immediate_commit_timestamp = struct.pack(
+            format_string, self.immediate_commit_timestamp
+        )
+
+        read_original_commit_timestamp = self.packet.read(7)
+        self.original_commit_timestamp = read_original_commit_timestamp.ljust(
+            struct.calcsize(format_string), b"\x00"
+        )
+        self.original_commit_timestamp = struct.unpack(
+            format_string, self.original_commit_timestamp
+        )[0]
+        self.original_commit_timestamp = self.original_commit_timestamp & (
+            (1 << (8 * struct.calcsize(format_string))) - 1
+        )
+        self.original_commit_timestamp = struct.pack(
+            format_string, self.original_commit_timestamp
+        )
+
+        # todo(hun): We should implement the properties below,
+        #            but due to the lack of documentation on the variable-length property,
+        #            we've left it as future work.
+        # - transaction_length, unknown Variable Length
+        # - immediate_server_version, 4 bytes
+        # - original_server_version, 4 bytes
+        # - Commit group ticket, 8 bytes, This property is only introduced in the CPP implementation.
+
     @property
     def gtid(self):
         """

pymysqlreplication/tests/base.py

@@ -74,6 +74,10 @@ def isMySQL57(self):
         version = float(self.getMySQLVersion().rsplit(".", 1)[0])
         return version == 5.7
 
+    def isMySQL57AndMore(self):
+        version = float(self.getMySQLVersion().rsplit(".", 1)[0])
+        return version >= 5.7
+
     def isMySQL80AndMore(self):
         version = float(self.getMySQLVersion().rsplit(".", 1)[0])
         return version >= 8.0

pymysqlreplication/tests/test_basic.py

@@ -14,7 +14,6 @@
 from pymysql.protocol import MysqlPacket
 import pytest
 
-
 __all__ = [
     "TestBasicBinLogStreamReader",
     "TestMultipleRowBinLogStreamReader",
@@ -1473,6 +1472,41 @@ def test_rows_query_log_event(self):
         self.assertIsInstance(event, RowsQueryLogEvent)
 
 
+class TestGtidEvent(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestGtidEvent, self).setUp()
+        self.execute("SET SESSION binlog_rows_query_log_events=1")
+
+    def tearDown(self):
+        self.execute("SET SESSION binlog_rows_query_log_events=0")
+        super(TestGtidEvent, self).tearDown()
+
+    def test_gtid_event(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[GtidEvent, FormatDescriptionEvent],
+        )
+        if not self.isMySQL57AndMore():
+            self.skipTest("Mysql version is under 5.7")
+        self.execute(
+            "CREATE TABLE IF NOT EXISTS test (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255))"
+        )
+        gtid_event = self.stream.fetchone()
+
+        self.assertIsInstance(gtid_event, GtidEvent)
+        self.assertIsInstance(gtid_event.event_type, int)
+        self.assertIsInstance(gtid_event.sid, bytes)
+        self.assertIsInstance(gtid_event.gno, int)
+        self.assertIsInstance(gtid_event.lt_type, int)
+        self.assertIsInstance(gtid_event.last_committed, int)
+        self.assertIsInstance(gtid_event.sequence_number, int)
+        self.assertEqual(gtid_event.sequence_number, 1)
+        self.assertIsInstance(gtid_event.immediate_commit_timestamp, bytes)
+        self.assertIsInstance(gtid_event.original_commit_timestamp, bytes)
+
+
 class TestLatin1(base.PyMySQLReplicationTestCase):
     def setUp(self):
         super().setUp(charset="latin1")