Fix byCode implementation

Author: shirolimit

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

@@ -70,7 +70,7 @@ public int getCode() {
     private static final ColumnType[] INDEX_BY_CODE;
 
     static {
-        INDEX_BY_CODE = new ColumnType[ColumnType.GEOMETRY.code];
+        INDEX_BY_CODE = new ColumnType[ColumnType.GEOMETRY.code + 1];
         for (ColumnType columnType : values()) {
             INDEX_BY_CODE[columnType.code] = columnType;
         }

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

@@ -569,9 +569,8 @@ public EventData deserialize(ByteArrayInputStream inputStream) throws IOExceptio
             public EventData deserialize(BinaryLogEventDataReader eventDataReader) throws IOException {
                 byte[] bytes = eventDataReader.readBytes(eventDataReader.available());
 
-                // TODO: use BinaryLogEventDataReader
-                EventData internalEventData = internal.deserialize(new ByteArrayInputStream(bytes));
-                EventData externalEventData = external.deserialize(new ByteArrayInputStream(bytes));
+                EventData internalEventData = internal.deserialize(new BinaryLogEventDataReader(bytes));
+                EventData externalEventData = external.deserialize(new BinaryLogEventDataReader(bytes));
                 return new EventDataWrapper(internalEventData, externalEventData);
             }
         }

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

@@ -405,7 +405,7 @@ private MetadataFieldType(int code) {
         private static final MetadataFieldType[] INDEX_BY_CODE;
 
         static {
-            INDEX_BY_CODE = new MetadataFieldType[MetadataFieldType.UNKNOWN_METADATA_FIELD_TYPE.code];
+            INDEX_BY_CODE = new MetadataFieldType[MetadataFieldType.UNKNOWN_METADATA_FIELD_TYPE.code + 1];
             for (MetadataFieldType fieldType : values()) {
                 INDEX_BY_CODE[fieldType.code] = fieldType;
             }

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

@@ -150,16 +150,14 @@ public TransactionPayloadEventData deserialize(BinaryLogEventDataReader eventDat
         byte[] dst = ByteBuffer.allocate(eventData.getUncompressedSize()).array();
         Zstd.decompressByteArray(dst, 0, dst.length, src, 0, src.length);
 
-        // Read and store events from decompressed byte array into input stream
         ArrayList<Event> decompressedEvents = new ArrayList<>();
         EventDeserializer transactionPayloadEventDeserializer = new EventDeserializer();
-        ByteArrayInputStream destinationInputStream = new ByteArrayInputStream(dst);
+        BinaryLogEventDataReader decompressedReader = new BinaryLogEventDataReader(dst);
 
-        // TODO: use BinaryLogEventDataReader here as well
-        Event internalEvent = transactionPayloadEventDeserializer.nextEvent(destinationInputStream);
+        Event internalEvent = transactionPayloadEventDeserializer.deserializeEvent(decompressedReader);
         while (internalEvent != null) {
             decompressedEvents.add(internalEvent);
-            internalEvent = transactionPayloadEventDeserializer.nextEvent(destinationInputStream);
+            internalEvent = transactionPayloadEventDeserializer.deserializeEvent(decompressedReader);
         }
 
         eventData.setUncompressedEvents(decompressedEvents);