Merge pull request #93 from noplay/speedup2

Speed up event processing

Author: julien-duponchelle

pymysqlreplication/packet.py

@@ -49,29 +49,30 @@ class BinLogPacketWrapper(object):
     }
 
     def __init__(self, from_packet, table_map, ctl_connection, use_checksum, allowed_events = None):
-        if not from_packet.is_ok_packet():
-            raise ValueError(
-                "Cannot create %s object from invalid packet type" %
-                self.__class__.__name__)
-
         # -1 because we ignore the ok byte
         self.read_bytes = 0
         # Used when we want to override a value in the data buffer
         self.__data_buffer = b''
 
-        # Ok Value
         self.packet = from_packet
-        self.packet.advance(1)
         self.charset = ctl_connection.charset
 
+        # OK value
+        # timestamp
+        # event_type
+        # server_id
+        # log_pos
+        # flags
+        unpack = struct.unpack('<cIcIIIH', self.packet.read(20))
+
         # Header
-        self.timestamp = struct.unpack('<I', self.packet.read(4))[0]
-        self.event_type = byte2int(self.packet.read(1))
-        self.server_id = struct.unpack('<I', self.packet.read(4))[0]
-        self.event_size = struct.unpack('<I', self.packet.read(4))[0]
+        self.timestamp = unpack[1]
+        self.event_type = byte2int(unpack[2])
+        self.server_id = unpack[3]
+        self.event_size = unpack[4]
         # position of the next event
-        self.log_pos = struct.unpack('<I', self.packet.read(4))[0]
-        self.flags = struct.unpack('<H', self.packet.read(2))[0]
+        self.log_pos = unpack[5]
+        self.flags = unpack[6]
 
         # MySQL 5.6 and more if binlog-checksum = CRC32
         if use_checksum:

pymysqlreplication/row_event.py

@@ -22,15 +22,16 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
         #Header
         self.table_id = self._read_table_id()
         self.primary_key = table_map[self.table_id].data["primary_key"]
-        self.flags = struct.unpack('<H', self.packet.read(2))[0]
 
         #Event V2
         if self.event_type == BINLOG.WRITE_ROWS_EVENT_V2 or \
                 self.event_type == BINLOG.DELETE_ROWS_EVENT_V2 or \
                 self.event_type == BINLOG.UPDATE_ROWS_EVENT_V2:
-                self.extra_data_length = struct.unpack('<H',
-                                                       self.packet.read(2))[0]
+                self.flags, self.extra_data_length = struct.unpack('<HH', self.packet.read(4))
                 self.extra_data = self.packet.read(self.extra_data_length / 8)
+        else:
+            self.flags = struct.unpack('<H', self.packet.read(2))[0]
+
         #Body
         self.number_of_columns = self.packet.read_length_coded_binary()
         self.columns = self.table_map[self.table_id].columns