add checksum detect in binlog file reader.

add some test code for checksum detection.

Author: zzt93

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

@@ -58,12 +58,13 @@ public BinaryLogFileReader(InputStream inputStream, EventDeserializer eventDeser
         if (eventDeserializer == null) {
             throw new IllegalArgumentException("Event deserializer cannot be NULL");
         }
-        this.inputStream = new ByteArrayInputStream(inputStream);
+        this.inputStream = new ByteArrayInputStream(new BufferedInputStream(inputStream));
         try {
             byte[] magicHeader = this.inputStream.read(MAGIC_HEADER.length);
             if (!Arrays.equals(magicHeader, MAGIC_HEADER)) {
                 throw new IOException("Not a valid binary log");
             }
+            this.eventDeserializer = eventDeserializer.detectChecksumType(this.inputStream);
         } catch (IOException e) {
             try {
                 this.inputStream.close();
@@ -72,7 +73,6 @@ public BinaryLogFileReader(InputStream inputStream, EventDeserializer eventDeser
             }
             throw e;
         }
-        this.eventDeserializer = eventDeserializer;
     }
 
     /**

src/main/java/com/github/shyiko/mysql/binlog/event/EventHeaderV4.java

@@ -21,6 +21,7 @@
  */
 public class EventHeaderV4 implements EventHeader {
 
+    public static final int HEADER_LEN = 19;
     // v1 (MySQL 3.23)
     private long timestamp;
     private EventType eventType;
@@ -87,7 +88,7 @@ public void setFlags(int flags) {
 
     @Override
     public long getHeaderLength() {
-        return 19;
+        return HEADER_LEN;
     }
 
     @Override

src/main/java/com/github/shyiko/mysql/binlog/event/FormatDescriptionEventData.java

@@ -23,6 +23,7 @@ public class FormatDescriptionEventData implements EventData {
     private int binlogVersion;
     private String serverVersion;
     private int headerLength;
+    private int eventLength;
 
     public int getBinlogVersion() {
         return binlogVersion;
@@ -48,6 +49,14 @@ public void setHeaderLength(int headerLength) {
         this.headerLength = headerLength;
     }
 
+    public void setEventLength(int eventLength) {
+        this.eventLength = eventLength;
+    }
+
+    public int getEventLength() {
+        return eventLength;
+    }
+
     @Override
     public String toString() {
         final StringBuilder sb = new StringBuilder();

src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java

@@ -18,7 +18,9 @@
 import com.github.shyiko.mysql.binlog.event.Event;
 import com.github.shyiko.mysql.binlog.event.EventData;
 import com.github.shyiko.mysql.binlog.event.EventHeader;
+import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
 import com.github.shyiko.mysql.binlog.event.EventType;
+import com.github.shyiko.mysql.binlog.event.FormatDescriptionEventData;
 import com.github.shyiko.mysql.binlog.event.TableMapEventData;
 import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
 
@@ -240,6 +242,27 @@ public EventDataDeserializer getEventDataDeserializer(EventType eventType) {
         return eventDataDeserializer != null ? eventDataDeserializer : defaultEventDataDeserializer;
     }
 
+    public EventDeserializer detectChecksumType(ByteArrayInputStream inputStream)
+        throws IOException {
+        inputStream.mark(EventHeaderV4.HEADER_LEN);
+        EventHeader eventHeader = eventHeaderDeserializer.deserialize(inputStream);
+        inputStream.reset();
+
+        inputStream.mark((int) (EventHeaderV4.HEADER_LEN + eventHeader.getDataLength()));
+        inputStream.skip(EventHeaderV4.HEADER_LEN); // skip header, not interpret it twice
+        FormatDescriptionEventDataDeserializer deserializer = new FormatDescriptionEventDataDeserializer();
+        FormatDescriptionEventData formatDescriptionEventData = deserializer.deserialize(inputStream);
+        int eventLength = formatDescriptionEventData.getEventLength();
+        // 1 byte means checksum algo, 4 byte is crc checksum content
+        if (eventLength + 1 + 4 == eventHeader.getDataLength()) {
+            setChecksumType(ChecksumType.CRC32);
+        } else {
+            setChecksumType(ChecksumType.NONE);
+        }
+        inputStream.reset();
+        return this;
+    }
+
     /**
      * @see CompatibilityMode#DATE_AND_TIME_AS_LONG
      * @see CompatibilityMode#DATE_AND_TIME_AS_LONG_MICRO

src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/FormatDescriptionEventDataDeserializer.java

@@ -15,6 +15,7 @@
  */
 package com.github.shyiko.mysql.binlog.event.deserialization;
 
+import com.github.shyiko.mysql.binlog.event.EventType;
 import com.github.shyiko.mysql.binlog.event.FormatDescriptionEventData;
 import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
 
@@ -32,7 +33,9 @@ public FormatDescriptionEventData deserialize(ByteArrayInputStream inputStream)
         eventData.setServerVersion(inputStream.readString(50).trim());
         inputStream.skip(4); // redundant, present in a header
         eventData.setHeaderLength(inputStream.readInteger(1));
-        // lengths for all event types
+        // skip lengths for some event types
+        inputStream.skip(EventType.FORMAT_DESCRIPTION.ordinal() - 1);
+        eventData.setEventLength(inputStream.readInteger(1));
         return eventData;
     }
 }

src/main/java/com/github/shyiko/mysql/binlog/io/ByteArrayInputStream.java

@@ -28,6 +28,8 @@ public class ByteArrayInputStream extends InputStream {
     private InputStream inputStream;
     private Integer peek;
     private int blockLength = -1;
+    private Integer peekMark;
+    private int blockMark;
 
     public ByteArrayInputStream(InputStream inputStream) {
         this.inputStream = inputStream;
@@ -218,4 +220,17 @@ public void skipToTheEndOfTheBlock() throws IOException {
         }
     }
 
+    @Override
+    public synchronized void mark(int readlimit) {
+        inputStream.mark(readlimit);
+        peekMark = peek;
+        blockMark = blockLength;
+    }
+
+    @Override
+    public synchronized void reset() throws IOException {
+        inputStream.reset();
+        peek = peekMark;
+        blockLength = blockMark;
+    }
 }

src/test/java/com/github/shyiko/mysql/binlog/BinaryLogFileReaderIntegrationTest.java

@@ -42,12 +42,32 @@ public class BinaryLogFileReaderIntegrationTest {
     public void testNextEvent() throws Exception {
         BinaryLogFileReader reader = new BinaryLogFileReader(new GZIPInputStream(
                 new FileInputStream("src/test/resources/mysql-bin.sakila.gz")));
+        readAll(reader, 1462);
+    }
+
+    @Test
+    public void testNoChecksum() throws Exception {
+        EventDeserializer eventDeserializer = new EventDeserializer();
+        BinaryLogFileReader reader = new BinaryLogFileReader(new GZIPInputStream(
+                new FileInputStream("src/test/resources/mysql-bin.sakila.gz")), eventDeserializer);
+        readAll(reader, 1462);
+    }
+
+    @Test
+    public void testCRCChecksum() throws Exception {
+        EventDeserializer eventDeserializer = new EventDeserializer();
+        BinaryLogFileReader reader = new BinaryLogFileReader(
+                new FileInputStream("/data/mysql-bin.000001"), eventDeserializer);
+        readAll(reader, 303);
+    }
+
+    private void readAll(BinaryLogFileReader reader, int expect) throws IOException {
         try {
             int numberOfEvents = 0;
             while ((reader.readEvent()) != null) {
                 numberOfEvents++;
             }
-            assertEquals(numberOfEvents, 1462);
+            assertEquals(numberOfEvents, expect);
         } finally {
             reader.close();
         }