Merge pull request julien-duponchelle#43 from noplay/handle_ctl_disconnects

julien-duponchelle · julien-duponchelle · commit d6394f6c2b7a · 2013-06-27T04:47:09.000-07:00
Handle disconnects for the ctl connection
diff --git a/pymysqlreplication/binlogstream.py b/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)
diff --git a/pymysqlreplication/row_event.py b/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
 
