fix: replication reconnect

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fix sudden loss of float precision. (#56)
 - Fix pool `echo` parameter not apply to create connection. (#62)
+- Fix replication reconnect.
 
 ### 0.2.7
 

Makefile

@@ -15,7 +15,6 @@ style: deps
 check: deps
 	@black --check $(checkfiles) || (echo "Please run 'make style' to auto-fix style issues" && false)
 	@ruff $(checkfiles)
-	@bandit -x tests -r $(checkfiles)
 	@mypy $(checkfiles)
 
 test: deps

asyncmy/replication/binlogstream.py

@@ -4,6 +4,7 @@
 from asyncmy import Connection
 from asyncmy.constants.COMMAND import COM_BINLOG_DUMP, COM_BINLOG_DUMP_GTID, COM_REGISTER_SLAVE
 from asyncmy.cursors import DictCursor
+from asyncmy.errors import OperationalError
 from asyncmy.replication.constants import (
     BINLOG_DUMP_NON_BLOCK,
     BINLOG_THROUGH_GTID,
@@ -93,6 +94,8 @@ def encoded(self, server_id: int, master_id: int = 0):
 
 
 class BinLogStream:
+    MYSQL_EXPECTED_ERROR_CODES = [2013, 2006]
+
     def __init__(
         self,
         connection: Connection,
@@ -275,8 +278,14 @@ async def close(self):
     async def _read(self):
         if not self._connected:
             await self._connect()
+        try:
+            pkt = await self._connection.read_packet()
+        except OperationalError as e:
+            code, _ = e.args
+            if code in self.MYSQL_EXPECTED_ERROR_CODES:
+                await self.close()
+                return
 
-        pkt = await self._connection.read_packet()
         if pkt.is_eof_packet():
             await self.close()
             raise StreamClosedError("BinLogStream is closed")

asyncmy/version.py

@@ -1 +1 @@
-__VERSION__ = "0.2.8-rc2"
+__VERSION__ = "0.2.8"