Fix bug replication and now don't need to connect manual

Author: long2ice

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Force int `pool_recycle`.
 - Fix `echo` option.
+- Fix bug replication and now don't need to connect manual.
 
 ### 0.1.8
 

README.md

@@ -105,7 +105,6 @@ async def run():
         resume_stream=True,
         blocking=True,
     )
-    await stream.connect()
     async for event in stream:
         print(event)
 
@@ -116,7 +115,7 @@ if __name__ == '__main__':
 
 ## ThanksTo
 
-> asyncmy is build on top of these nice projects.
+> asyncmy is build on top of these awesome projects.
 
 - [pymysql](https://github/pymysql/PyMySQL), a pure python MySQL client.
 - [aiomysql](https://github.com/aio-libs/aiomysql), a library for accessing a MySQL database from the asyncio.

asyncmy/replication/binlogstream.py

@@ -179,9 +179,7 @@ def _allowed_event_list(
                 pass
         return frozenset(events)
 
-    async def connect(self):
-        if self._connected:
-            return
+    async def _connect(self):
         await self._connection.connect()
         self._use_checksum = await self._checksum_enable()
         async with self._connection.cursor() as cursor:
@@ -272,7 +270,7 @@ async def close(self):
 
     async def _read(self):
         if not self._connected:
-            await self.connect()
+            await self._connect()
 
         pkt = await self._connection.read_packet()
         if pkt.is_eof_packet():
@@ -316,7 +314,7 @@ async def _read(self):
     async def _checksum_enable(self):
         async with self._connection.cursor() as cursor:
             await cursor.execute("SHOW GLOBAL VARIABLES LIKE 'BINLOG_CHECKSUM'")
-            result = cursor.fetchone()
+            result = await cursor.fetchone()
             if result is None:
                 return False
             var, value = result[:2]
@@ -354,6 +352,8 @@ def __aiter__(self):
         return self
 
     async def __anext__(self):
+        if not self._connected:
+            await self._connect()
         try:
             ret = await self._read()
             while ret is None:

asyncmy/replication/events.py

@@ -1,6 +1,8 @@
 import binascii
 import struct
 
+from asyncmy.replication.utils import byte2int, int2byte
+
 
 class BinLogEvent:
     def __init__(
@@ -35,7 +37,7 @@ def processed(self):
     def _read_table_id(self):
         # Table ID is 6 byte
         # pad little-endian number
-        table_id = self.packet.read(6) + bytes(0) + bytes(0)
+        table_id = self.packet.read(6) + int2byte(0) + int2byte(0)
         return struct.unpack("<Q", table_id)[0]
 
     async def init(self):
@@ -127,7 +129,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         # Post-header
         self.slave_proxy_id = self.packet.read_uint32()
         self.execution_time = self.packet.read_uint32()
-        self.schema_length = self.packet.read(1)
+        self.schema_length = byte2int(self.packet.read(1))
         self.error_code = self.packet.read_uint16()
         self.status_vars_length = self.packet.read_uint16()
 

asyncmy/replication/packets.py

@@ -48,6 +48,7 @@
     WRITE_ROWS_EVENT_V2,
     XID_EVENT,
 )
+from asyncmy.replication.utils import byte2int
 
 
 class BinLogPacket:
@@ -99,7 +100,7 @@ def __init__(
         # server_id
         # log_pos
         # flags
-        unpack = struct.unpack("<cIcIIIH", self._packet.read(20))
+        unpack = struct.unpack("<cIBIIIH", self._packet.read(20))
 
         # Header
         self.timestamp = unpack[1]
@@ -166,7 +167,7 @@ def advance(self, size):
             self._packet.advance(size)
 
     def read_length_coded_binary(self):
-        c = self.read(1)
+        c = byte2int(self.read(1))
         if c == NULL_COLUMN:
             return None
         if c < UNSIGNED_CHAR_COLUMN:

asyncmy/replication/row_events.py

@@ -40,6 +40,7 @@
 from .errors import TableMetadataUnavailableError
 from .events import BinLogEvent
 from .table import Table
+from .utils import byte2int
 
 
 class RowsEvent(BinLogEvent):
@@ -539,10 +540,10 @@ def __init__(self, from_packet, event_size, table_map, connection, **kwargs):
         self.flags = struct.unpack("<H", self.packet.read(2))[0]
 
         # Payload
-        self.schema_length = self.packet.read(1)
+        self.schema_length = byte2int(self.packet.read(1))
         self.schema = self.packet.read(self.schema_length).decode()
         self.packet.advance(1)
-        self.table_length = self.packet.read(1)
+        self.table_length = byte2int(self.packet.read(1))
         self.table_name = self.packet.read(self.table_length).decode()
         schema_table = f"{self.schema}.{self.table_name}"
         if self._only_tables is not None and schema_table not in self._only_tables:

asyncmy/replication/utils.py

@@ -0,0 +1,12 @@
+import struct
+
+
+def byte2int(b):
+    if isinstance(b, int):
+        return b
+    else:
+        return struct.unpack("!B", b)[0]
+
+
+def int2byte(i: int):
+    return struct.pack("!B", i)

tests/test_replication.py renamed to examples/replication.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 from asyncmy import connect
@@ -6,21 +8,23 @@
 from conftest import connection_kwargs
 
 
-@pytest.mark.skip(reason="need test in local")
-@pytest.mark.asyncio
-async def test_binlogstream(connection):
+async def main():
     conn = await connect(**connection_kwargs)
+    ctl_conn = await connect(**connection_kwargs)
 
     stream = BinLogStream(
-        connection,
         conn,
+        ctl_conn,
         1,
-        master_log_file="binlog.000172",
+        master_log_file="binlog.000019",
+        master_log_position=155276,
         resume_stream=True,
         blocking=True,
-        master_log_position=2235312,
     )
-    await stream.connect()
     async for event in stream:
         if isinstance(event, WriteRowsEvent):
             print(event.schema, event.table, event.rows)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())