Get mysql version from FormatDescriptionEvent, and it's used while parsing GtidEvent

Author: etern

pymysqlreplication/binlogstream.py

@@ -8,7 +8,7 @@
 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,
@@ -211,6 +211,7 @@ def __init__(self, connection_settings, server_id, ctl_connection_settings=None,
             self.pymysql_wrapper = pymysql_wrapper
         else:
             self.pymysql_wrapper = pymysql.connect
+        self.mysql_version = (0, 0, 0)
 
     def close(self):
         if self.__connected_stream:
@@ -428,6 +429,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,
@@ -490,6 +492,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,

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
@@ -59,8 +61,10 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         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]
+
+        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):
@@ -76,8 +80,9 @@ 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)
+        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
@@ -135,7 +140,17 @@ def _dump(self):
 
 
 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

@@ -86,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,
@@ -132,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,