Handle disconnects for the ctl connection

The problem is that this module should run for a long time. Mysql disconnects long running connections and therefore getting table information causes the whole program to crash.

Author: bjoernhaeuser

pymysqlreplication/binlogstream.py

@@ -57,6 +57,7 @@ def __connect_to_ctl(self):
  self._ctl_connection_settings["cursorclass"] = \
  pymysql.cursors.DictCursor
  self._ctl_connection = pymysql.connect(**self._ctl_connection_settings)
+ self._ctl_connection._get_table_information = self.__get_table_information
  self.__connected_ctl = True
 
  def __connect_to_stream(self):
@@ -117,8 +118,7 @@ def fetchone(self):
  if not pkt.is_ok_packet():
  continue
 
- binlog_event = BinLogPacketWrapper(pkt, self.table_map,
- self._ctl_connection)
+ binlog_event = BinLogPacketWrapper(pkt, self.table_map, self._ctl_connection)
 
  if binlog_event.event_type == TABLE_MAP_EVENT:
  self.table_map[binlog_event.event.table_id] = \
@@ -146,5 +146,32 @@ def __filter_event(self, event):
  return True
  return False
 
+ def __get_table_information(self, schema, table):
+ for i in range(1, 3):
+ try:
+ if not self.__connected_ctl:
+ self.__connect_to_ctl()
+
+ cur = self._ctl_connection.cursor()
+ cur.execute("""
+ SELECT
+ COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME,
+ COLUMN_COMMENT, COLUMN_TYPE
+ FROM
+ columns
+ WHERE
+ table_schema = %s AND table_name = %s
+ """, (schema, table))
+
+ return cur.fetchall()
+ except pymysql.OperationalError as error:
+ code, message = error.args
+ # 2013: Connection Lost
+ if code == 2013:
+ self.__connected_ctl = False
+ continue
+ else:
+ raise error
+
  def __iter__(self):
  return iter(self.fetchone, None)

pymysqlreplication/row_event.py

@@ -477,8 +477,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
  if self.table_id in table_map:
  self.column_schemas = table_map[self.table_id].column_schemas
  else:
- self.column_schemas = self.__get_table_information(self.schema,
- self.table)
+ self.column_schemas = self._ctl_connection._get_table_information(self.schema, self.table)
 
  # Read columns meta data
  column_types = list(self.packet.read(self.column_count))
@@ -495,19 +494,6 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection):
  # TODO: get this informations instead of trashing data
  # n NULL-bitmask, length: (column-length * 8) / 7
 
- def __get_table_information(self, schema, table):
- cur = self._ctl_connection.cursor()
- cur.execute("""
- SELECT
- COLUMN_NAME, COLLATION_NAME, CHARACTER_SET_NAME,
- COLUMN_COMMENT, COLUMN_TYPE
- FROM
- columns
- WHERE
- table_schema = %s AND table_name = %s
- """, (schema, table))
- return cur.fetchall()
-
  def get_table(self):
  return self.table_obj