Fixed handling of packets exceeding 16mb in size

Author: shyiko

src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java

@@ -35,6 +35,7 @@
 import com.github.shyiko.mysql.binlog.network.SocketFactory;
 import com.github.shyiko.mysql.binlog.network.protocol.ErrorPacket;
 import com.github.shyiko.mysql.binlog.network.protocol.GreetingPacket;
+import com.github.shyiko.mysql.binlog.network.protocol.Packet;
 import com.github.shyiko.mysql.binlog.network.protocol.PacketChannel;
 import com.github.shyiko.mysql.binlog.network.protocol.ResultSetRowPacket;
 import com.github.shyiko.mysql.binlog.network.protocol.command.AuthenticateCommand;
@@ -597,7 +598,9 @@ private void listenForEventPackets() throws IOException {
                 }
                 Event event;
                 try {
-                    event = eventDeserializer.nextEvent(inputStream);
+                    event = eventDeserializer.nextEvent(packetLength == Packet.MAX_LENGTH ?
+                        new ByteArrayInputStream(readPacketSplitInChunks(inputStream, packetLength - 1)) :
+                        inputStream);
                 } catch (Exception e) {
                     Throwable cause = e instanceof EventDataDeserializationException ? e.getCause() : e;
                     if (cause instanceof EOFException || cause instanceof SocketException) {
@@ -633,6 +636,25 @@ private void listenForEventPackets() throws IOException {
         }
     }
 
+    private byte[] readPacketSplitInChunks(ByteArrayInputStream inputStream, int packetLength) throws IOException {
+        List<byte[]> chunks = new LinkedList<byte[]>();
+        chunks.add(inputStream.read(packetLength));
+        int chunkLength;
+        do {
+            chunkLength = inputStream.readInteger(3);
+            inputStream.skip(1); // 1 byte for sequence
+            chunks.add(inputStream.read(chunkLength));
+            packetLength += chunkLength;
+        } while (chunkLength == Packet.MAX_LENGTH);
+        byte[] buffer = new byte[packetLength];
+        int offset = 0;
+        for (byte[] chunk : chunks) {
+            System.arraycopy(chunk, 0, buffer, offset, chunk.length);
+            offset += chunk.length;
+        }
+        return buffer;
+    }
+
     private void updateClientBinlogFilenameAndPosition(Event event) {
         EventHeader eventHeader = event.getHeader();
         if (eventHeader.getEventType() == EventType.ROTATE) {

src/main/java/com/github/shyiko/mysql/binlog/network/protocol/Packet.java

@@ -20,4 +20,6 @@
  */
 public interface Packet {
 
+    // https://dev.mysql.com/doc/internals/en/sending-more-than-16mbyte.html
+    int MAX_LENGTH = 16777215;
 }