Merge remote-tracking branch 'origin/master'

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -8,11 +8,11 @@
 from pymysql.cursors import DictCursor
 
 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, MariadbGtidEvent)
 from .exceptions import BinLogNotEnabled
@@ -232,6 +232,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:
@@ -505,6 +506,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,
@@ -572,6 +574,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,
@@ -584,6 +589,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,
@@ -22,6 +23,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
         self._ignore_decode_errors = ignore_decode_errors
         # The event have been fully processed, if processed is false
@@ -60,6 +62,11 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.commit_flag = struct.unpack("!B", self.packet.read(1))[0] == 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):
@@ -75,6 +82,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
@@ -121,8 +131,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

@@ -69,6 +69,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,
@@ -89,7 +90,10 @@ class BinLogPacketWrapper(object):
         constants.MARIADB_START_ENCRYPTION_EVENT: event.NotImplementedEvent
     }
 
-    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,
@@ -136,6 +140,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,