Merge pull request julien-duponchelle#257 from etern/master

Add XAPrepareEvent, parse last_committed & sequence_number of GtidEvent

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -9,11 +9,11 @@
 from pymysql.util import int2byte
 
 from .packet import BinLogPacketWrapper
-from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT
+from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT, FORMAT_DESCRIPTION_EVENT
 from .gtid import GtidSet
 from .event import (
     QueryEvent, RotateEvent, FormatDescriptionEvent,
-    XidEvent, GtidEvent, StopEvent,
+    XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent)
 from .exceptions import BinLogNotEnabled
@@ -219,6 +219,7 @@ def __init__(self, connection_settings, server_id,
             self.pymysql_wrapper = pymysql_wrapper
         else:
             self.pymysql_wrapper = pymysql.connect
+        self.mysql_version = (0, 0, 0)
 
     def close(self):
         if self.__connected_stream:
@@ -446,6 +447,7 @@ def fetchone(self):
 
             binlog_event = BinLogPacketWrapper(pkt, self.table_map,
                                                self._ctl_connection,
+                                               self.mysql_version,
                                                self.__use_checksum,
                                                self.__allowed_events_in_packet,
                                                self.__only_tables,
@@ -508,6 +510,9 @@ def fetchone(self):
             if binlog_event.event is None or (binlog_event.event.__class__ not in self.__allowed_events):
                 continue
 
+            if binlog_event.event_type == FORMAT_DESCRIPTION_EVENT:
+                self.mysql_version = binlog_event.event.mysql_version
+
             return binlog_event.event
 
     def _allowed_event_list(self, only_events, ignored_events,
@@ -520,6 +525,7 @@ def _allowed_event_list(self, only_events, ignored_events,
                 RotateEvent,
                 StopEvent,
                 FormatDescriptionEvent,
+                XAPrepareEvent,
                 XidEvent,
                 GtidEvent,
                 BeginLoadQueryEvent,

pymysqlreplication/constants/BINLOG.py

@@ -36,6 +36,7 @@
 GTID_LOG_EVENT = 0x21
 ANONYMOUS_GTID_LOG_EVENT = 0x22
 PREVIOUS_GTIDS_LOG_EVENT = 0x23
+XA_PREPARE_EVENT = 0x26
 
 # INTVAR types
 INTVAR_INVALID_INT_EVENT = 0x00

pymysqlreplication/event.py

@@ -9,6 +9,7 @@
 
 class BinLogEvent(object):
     def __init__(self, from_packet, event_size, table_map, ctl_connection,
+                 mysql_version=(0,0,0),
                  only_tables=None,
                  ignored_tables=None,
                  only_schemas=None,
@@ -21,6 +22,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection,
         self.timestamp = self.packet.timestamp
         self.event_size = event_size
         self._ctl_connection = ctl_connection
+        self.mysql_version = mysql_version
         self._fail_on_table_metadata_unavailable = fail_on_table_metadata_unavailable
         # The event have been fully processed, if processed is false
         # the event will be skipped
@@ -58,6 +60,11 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.commit_flag = byte2int(self.packet.read(1)) == 1
         self.sid = self.packet.read(16)
         self.gno = struct.unpack('<Q', self.packet.read(8))[0]
+        self.lt_type = byte2int(self.packet.read(1))
+
+        if self.mysql_version >= (5, 7):
+            self.last_committed = struct.unpack('<Q', self.packet.read(8))[0]
+            self.sequence_number = struct.unpack('<Q', self.packet.read(8))[0]
 
     @property
     def gtid(self):
@@ -73,6 +80,9 @@ def gtid(self):
     def _dump(self):
         print("Commit: %s" % self.commit_flag)
         print("GTID_NEXT: %s" % self.gtid)
+        if hasattr(self, "last_committed"):
+            print("last_committed: %d" % self.last_committed)
+            print("sequence_number: %d" % self.sequence_number)
 
     def __repr__(self):
         return '<GtidEvent "%s">' % self.gtid
@@ -98,8 +108,49 @@ def dump(self):
         print()
 
 
+class XAPrepareEvent(BinLogEvent):
+    """An XA prepare event is generated for a XA prepared transaction.
+    Like Xid_event it contans XID of the *prepared* transaction
+
+    Attributes:
+        one_phase: current XA transaction commit method
+        xid: serialized XID representation of XA transaction
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(XAPrepareEvent, self).__init__(from_packet, event_size, table_map,
+                                          ctl_connection, **kwargs)
+
+        # one_phase is True: XA COMMIT ... ONE PHASE
+        # one_phase is False: XA PREPARE
+        self.one_phase = (self.packet.read(1) != b'\x00')
+        self.xid_format_id = struct.unpack('<I', self.packet.read(4))[0]
+        gtrid_length = struct.unpack('<I', self.packet.read(4))[0]
+        bqual_length = struct.unpack('<I', self.packet.read(4))[0]
+        self.xid_gtrid = self.packet.read(gtrid_length)
+        self.xid_bqual = self.packet.read(bqual_length)
+
+    @property
+    def xid(self):
+        return self.xid_gtrid.decode() + self.xid_bqual.decode()
+
+    def _dump(self):
+        print("One phase: %s" % self.one_phase)
+        print("XID formatID: %d" % self.xid_format_id)
+        print("XID: %s" % self.xid)
+
+
 class FormatDescriptionEvent(BinLogEvent):
-    pass
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(FormatDescriptionEvent, self).__init__(from_packet, event_size, table_map,
+                                          ctl_connection, **kwargs)
+        self.binlog_version = struct.unpack('<H', self.packet.read(2))
+        self.mysql_version_str = self.packet.read(50).rstrip(b'\0').decode()
+        numbers = self.mysql_version_str.split('-')[0]
+        self.mysql_version = tuple(map(int, numbers.split('.')))
+
+    def _dump(self):
+        print("Binlog version: %s" % self.binlog_version)
+        print("MySQL version: %s" % self.mysql_version_str)
 
 
 class StopEvent(BinLogEvent):

pymysqlreplication/packet.py

@@ -71,6 +71,7 @@ class BinLogPacketWrapper(object):
         constants.BEGIN_LOAD_QUERY_EVENT: event.BeginLoadQueryEvent,
         constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
         constants.HEARTBEAT_LOG_EVENT: event.HeartbeatLogEvent,
+        constants.XA_PREPARE_EVENT: event.XAPrepareEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
         constants.WRITE_ROWS_EVENT_V1: row_event.WriteRowsEvent,
@@ -85,7 +86,10 @@ class BinLogPacketWrapper(object):
 
     }
 
-    def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
+    def __init__(self, from_packet, table_map,
+                 ctl_connection,
+                 mysql_version,
+                 use_checksum,
                  allowed_events,
                  only_tables,
                  ignored_tables,
@@ -131,6 +135,7 @@ def __init__(self, from_packet, table_map, ctl_connection, use_checksum,
             return
         self.event = event_class(self, event_size_without_header, table_map,
                                  ctl_connection,
+                                 mysql_version=mysql_version,
                                  only_tables=only_tables,
                                  ignored_tables=ignored_tables,
                                  only_schemas=only_schemas,

pymysqlreplication/tests/test_basic.py

@@ -25,9 +25,9 @@ def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 14)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 13)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 13)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 15)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 14)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 14)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):