Add XAPrepareEvent, parse last_committed & sequence_number of GtidEvent

Author: etern

pymysqlreplication/binlogstream.py

@@ -12,7 +12,7 @@
 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
@@ -502,6 +502,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

@@ -58,6 +58,9 @@ 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))
+        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 +76,8 @@ def gtid(self):
     def _dump(self):
         print("Commit: %s" % self.commit_flag)
         print("GTID_NEXT: %s" % self.gtid)
+        print("last_committed: %d" % self.last_committed)
+        print("sequence_number: %d" % self.sequence_number)
 
     def __repr__(self):
         return '<GtidEvent "%s">' % self.gtid
@@ -98,6 +103,37 @@ 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
 

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,