Merge branch 'pr/21'

Author: osheroff

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

@@ -75,6 +75,7 @@ public abstract class AbstractRowsEventDataDeserializer<T extends EventData> imp
     private Long invalidDateAndTimeRepresentation;
     private boolean microsecondsPrecision;
     private boolean deserializeCharAndBinaryAsByteArray;
+    private boolean deserializeIntegerAsByteArray;
 
     public AbstractRowsEventDataDeserializer(Map<Long, TableMapEventData> tableMapEventByTableId) {
         this.tableMapEventByTableId = tableMapEventByTableId;
@@ -97,6 +98,10 @@ void setDeserializeCharAndBinaryAsByteArray(boolean value) {
         this.deserializeCharAndBinaryAsByteArray = value;
     }
 
+    void setDeserializeIntegerAsByteArray(boolean deserializeIntegerAsByteArray) {
+        this.deserializeIntegerAsByteArray = deserializeIntegerAsByteArray;
+    }
+
     protected Serializable[] deserializeRow(long tableId, BitSet includedColumns, ByteArrayInputStream inputStream)
             throws IOException {
         TableMapEventData tableMapEvent = tableMapEventByTableId.get(tableId);
@@ -203,22 +208,37 @@ protected Serializable deserializeBit(int meta, ByteArrayInputStream inputStream
     }
 
     protected Serializable deserializeTiny(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(1);
+        }
         return (int) ((byte) inputStream.readInteger(1));
     }
 
     protected Serializable deserializeShort(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(2);
+        }
         return (int) ((short) inputStream.readInteger(2));
     }
 
     protected Serializable deserializeInt24(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(3);
+        }
         return (inputStream.readInteger(3) << 8) >> 8;
     }
 
     protected Serializable deserializeLong(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(4);
+        }
         return inputStream.readInteger(4);
     }
 
     protected Serializable deserializeLongLong(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(8);
+        }
         return inputStream.readLong(8);
     }
 

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

@@ -200,6 +200,9 @@ private void ensureCompatibility(EventDataDeserializer eventDataDeserializer) {
             deserializer.setDeserializeCharAndBinaryAsByteArray(
                 compatibilitySet.contains(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY)
             );
+            deserializer.setDeserializeIntegerAsByteArray(
+                compatibilitySet.contains(CompatibilityMode.INTEGER_AS_BYTE_ARRAY)
+            );
         }
     }
 
@@ -350,7 +353,11 @@ public enum CompatibilityMode {
          *
          * <p>This option is going to be enabled by default starting from mysql-binlog-connector-java@1.0.0.
          */
-        CHAR_AND_BINARY_AS_BYTE_ARRAY
+        CHAR_AND_BINARY_AS_BYTE_ARRAY,
+        /**
+         * Return TINY/SHORT/INT24/LONG/LONGLONG values as byte[]|s (instead of int|s).
+         */
+        INTEGER_AS_BYTE_ARRAY
     }
 
     /**

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

@@ -421,6 +421,51 @@ public void testDeserializationOfDateAndTimeAsLong() throws Exception {
         }
     }
 
+    @Test
+    public void testDeserializationOfIntegerAsByteArray() throws Exception {
+        final BinaryLogClient client = new BinaryLogClient(slave.hostname, slave.port,
+            slave.username, slave.password);
+        EventDeserializer eventDeserializer = new EventDeserializer();
+        eventDeserializer.setCompatibilityMode(CompatibilityMode.INTEGER_AS_BYTE_ARRAY);
+        client.setEventDeserializer(eventDeserializer);
+        client.connect(DEFAULT_TIMEOUT);
+        try {
+            Serializable[] result;
+
+            result = writeAndCaptureRow("tinyint unsigned", "0", "1", "255");
+            assertEquals(result[0], 0);
+            assertEquals(result[1], 1);
+            assertEquals(result[2], -1);
+
+
+            result = writeAndCaptureRow("tinyint", "-128", "-1", "0", "1", "127");
+            assertEquals(result[0], -128);
+            assertEquals(result[1], -1);
+            assertEquals(result[2], 0);
+            assertEquals(result[3], 1);
+            assertEquals(result[4], 127);
+
+            result = writeAndCaptureRow("smallint unsigned", "0", "1", "65535");
+            assertEquals(result[0], 0);
+            assertEquals(result[1], 1);
+            assertEquals(result[2], -1);
+
+            result = writeAndCaptureRow("smallint", "-32768", "-1", "0", "1", "32767");
+            assertEquals(result[0], -32768);
+            assertEquals(result[1], -1);
+            assertEquals(result[2], 0);
+            assertEquals(result[3], 1);
+            assertEquals(result[4], 32767);
+
+            result = writeAndCaptureRow("mediumint unsigned", "0", "1", "16777215");
+            assertEquals(result[0], 0);
+            assertEquals(result[1], 1);
+            assertEquals(result[2], -1);
+        } finally {
+            client.disconnect();
+        }
+    }
+
     @Test
     public void testDeserializationOfDateAndTimeAsLongMicrosecondsPrecision() throws Exception {
         final BinaryLogClient client = new BinaryLogClient(slave.hostname, slave.port,