Custom Row value codec

vertx-mysql-client/src/main/java/examples/SqlClientExamples.java

@@ -16,11 +16,15 @@
  */
 package examples;
 
+import io.netty.buffer.ByteBuf;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.docgen.Source;
 import io.vertx.mysqlclient.MySQLConnectOptions;
+import io.vertx.mysqlclient.typecodec.MySQLDataTypeDefaultCodecs;
+import io.vertx.mysqlclient.typecodec.MySQLType;
 import io.vertx.sqlclient.Cursor;
 import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.PoolOptions;
@@ -33,7 +37,13 @@
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.Transaction;
 import io.vertx.sqlclient.Tuple;
+import io.vertx.sqlclient.codec.DataType;
+import io.vertx.sqlclient.codec.DataTypeCodec;
+import io.vertx.sqlclient.codec.DataTypeCodecRegistry;
+import io.vertx.sqlclient.impl.codec.CommonCodec;
 
+import java.nio.charset.Charset;
+import java.sql.JDBCType;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -328,4 +338,164 @@ public void usingCursors03(SqlConnection connection) {
   public void tracing01(MySQLConnectOptions options) {
     options.setTracingPolicy(TracingPolicy.ALWAYS);
   }
+
+  public void customDataTypeCodecExample01(SqlConnection connection) {
+    // usecase01-byte as boolean
+    DataTypeCodec<Boolean, Boolean> tinyintCodec = new DataTypeCodec<Boolean, Boolean>() {
+
+      private final DataType<Boolean, Boolean> tinyintDatatype = new DataType<Boolean, Boolean>() {
+        @Override
+        public int identifier() {
+          return MySQLType.TINYINT.identifier();
+        }
+
+        @Override
+        public JDBCType jdbcType() {
+          return JDBCType.TINYINT;
+        }
+
+        @Override
+        public Class<Boolean> encodingJavaClass() {
+          return Boolean.class;
+        }
+
+        @Override
+        public Class<Boolean> decodingJavaClass() {
+          return Boolean.class;
+        }
+      };
+
+      @Override
+      public DataType<Boolean, Boolean> dataType() {
+        return tinyintDatatype;
+      }
+
+      @Override
+      public void encode(ByteBuf buffer, Boolean value) {
+        buffer.writeBoolean(value);
+      }
+
+      @Override
+      public Boolean binaryDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        return buffer.readBoolean();
+      }
+
+      @Override
+      public Boolean textualDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        byte b = (byte) CommonCodec.decodeDecStringToLong(readerIndex, (int) length, buffer);
+        return b != 0;
+      }
+    };
+
+    DataTypeCodecRegistry dataTypeCodecRegistry = connection.dataTypeCodecRegistry();
+    dataTypeCodecRegistry.unregister(MySQLDataTypeDefaultCodecs.TinyIntTypeCodec.INSTANCE);
+    dataTypeCodecRegistry.register(tinyintCodec);
+  }
+
+  public void customDataTypeCodecExample02(SqlConnection connection) {
+    // usecase02-custom JSON https://github.com/eclipse-vertx/vertx-sql-client/issues/862
+    DataTypeCodec<CustomJSON, CustomJSON> jsonCustomTypeCodec = new DataTypeCodec<CustomJSON, CustomJSON>() {
+
+      private final DataType<CustomJSON, CustomJSON> customJSONDatatype = new DataType<CustomJSON, CustomJSON>() {
+        @Override
+        public int identifier() {
+          return MySQLType.JSON.identifier();
+        }
+
+        @Override
+        public JDBCType jdbcType() {
+          return JDBCType.OTHER;
+        }
+
+        @Override
+        public Class<CustomJSON> encodingJavaClass() {
+          return CustomJSON.class;
+        }
+
+        @Override
+        public Class<CustomJSON> decodingJavaClass() {
+          return CustomJSON.class;
+        }
+      };
+
+      @Override
+      public DataType<CustomJSON, CustomJSON> dataType() {
+        return customJSONDatatype;
+      }
+
+      @Override
+      public void encode(ByteBuf buffer, CustomJSON value) {
+        // write to bytebuf directly from the custom json
+        // ...
+      }
+
+      @Override
+      public CustomJSON binaryDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        // read from bytebuf directly
+        // ...
+      }
+
+      @Override
+      public CustomJSON textualDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        // read from bytebuf directly
+        // ...
+      }
+    };
+
+    DataTypeCodecRegistry dataTypeCodecRegistry = connection.dataTypeCodecRegistry();
+    dataTypeCodecRegistry.unregister(MySQLDataTypeDefaultCodecs.JsonTypeCodec.INSTANCE);
+    dataTypeCodecRegistry.register(jsonCustomTypeCodec);
+  }
+
+  public void customDataTypeCodecExample03(SqlConnection connection) {
+    // usecase03 - row values direct buffer
+    DataTypeCodec<ByteBuf, ByteBuf> rowValueRawBufCodec = new DataTypeCodec<ByteBuf, ByteBuf>() {
+
+      private final DataType<ByteBuf, ByteBuf> tinyintDatatype = new DataType<ByteBuf, ByteBuf>() {
+        @Override
+        public int identifier() {
+          return MySQLType.TINYINT.identifier();
+        }
+
+        @Override
+        public JDBCType jdbcType() {
+          return JDBCType.TINYINT;
+        }
+
+        @Override
+        public Class<ByteBuf> encodingJavaClass() {
+          return ByteBuf.class;
+        }
+
+        @Override
+        public Class<ByteBuf> decodingJavaClass() {
+          return ByteBuf.class;
+        }
+      };
+
+      @Override
+      public DataType<ByteBuf, ByteBuf> dataType() {
+        return tinyintDatatype;
+      }
+
+      @Override
+      public void encode(ByteBuf buffer, ByteBuf value) {
+        buffer.writeBytes(value);
+      }
+
+      @Override
+      public ByteBuf binaryDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        return buffer.readRetainedSlice((int) length);
+      }
+
+      @Override
+      public ByteBuf textualDecode(ByteBuf buffer, int readerIndex, long length, Charset charset) {
+        return buffer.readRetainedSlice((int) length);
+      }
+    };
+
+    DataTypeCodecRegistry dataTypeCodecRegistry = connection.dataTypeCodecRegistry();
+    dataTypeCodecRegistry.unregister(MySQLDataTypeDefaultCodecs.TinyIntTypeCodec.INSTANCE);
+    dataTypeCodecRegistry.register(rowValueRawBufCodec);
+  }
 }

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLRowImpl.java

@@ -14,7 +14,6 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.mysqlclient.data.spatial.*;
-import io.vertx.mysqlclient.impl.datatype.DataType;
 import io.vertx.mysqlclient.impl.protocol.ColumnDefinition;
 import io.vertx.sqlclient.Row;
 import io.vertx.sqlclient.data.Numeric;
@@ -246,7 +245,7 @@ private GeometryCollection getGeometryCollection(int pos) {
   public LocalTime getLocalTime(int pos) {
     ColumnDefinition columnDefinition = rowDesc.columnDefinitions()[pos];
     Object val = getValue(pos);
-    if (columnDefinition.type() == DataType.TIME && val instanceof Duration) {
+    if (columnDefinition.type() == ColumnDefinition.ColumnType.MYSQL_TYPE_TIME && val instanceof Duration) {
       // map MySQL TIME data type to java.time.LocalTime
       Duration duration = (Duration) val;
       return LocalTime.ofNanoOfDay(duration.toNanos());

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java

@@ -27,6 +27,8 @@
 import io.vertx.mysqlclient.SslMode;
 import io.vertx.mysqlclient.impl.codec.MySQLCodec;
 import io.vertx.mysqlclient.impl.command.InitialHandshakeCommand;
+import io.vertx.mysqlclient.typecodec.MySQLDataTypeCodecRegistry;
+import io.vertx.sqlclient.codec.DataTypeCodecRegistry;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.QueryResultHandler;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
@@ -103,4 +105,9 @@ public void upgradeToSsl(Handler<AsyncResult<Void>> completionHandler) {
   public DatabaseMetadata getDatabaseMetaData() {
     return metaData;
   }
+
+  @Override
+  public MySQLDataTypeCodecRegistry getDataTypeCodecRegistry() {
+    return MySQLDataTypeCodecRegistry.INSTANCE;
+  }
 }

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/CommandCodec.java

@@ -175,7 +175,7 @@ ColumnDefinition decodeColumnDefinitionPacketPayload(ByteBuf payload) {
     long lengthOfFixedLengthFields = BufferUtils.readLengthEncodedInteger(payload);
     int characterSet = payload.readUnsignedShortLE();
     long columnLength = payload.readUnsignedIntLE();
-    DataType type = DataType.valueOf(payload.readUnsignedByte());
+    short type = payload.readUnsignedByte();
     int flags = payload.readUnsignedShortLE();
     byte decimals = payload.readByte();
     return new ColumnDefinition(catalog, schema, table, orgTable, name, orgName, characterSet, columnLength, type, flags, decimals);

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/ExtendedBatchQueryCommandCodec.java

@@ -106,6 +106,7 @@ private void sendBatchStatementExecuteCommand(MySQLPreparedStatement statement,
       for (int i = 0; i < numOfParams; i++) {
         Object value = params.getValue(i);
         if (value != null) {
+          // FIXME encoding using customized codec
           DataTypeCodec.encodeBinary(DataTypeCodec.inferDataTypeByEncodingValue(value), value, encoder.encodingCharset, packet);
         } else {
           nullBitmap[i / 8] |= (1 << (i & 7));

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/ExtendedQueryCommandCodec.java

@@ -20,6 +20,7 @@
 import io.vertx.mysqlclient.impl.datatype.DataType;
 import io.vertx.mysqlclient.impl.datatype.DataTypeCodec;
 import io.vertx.mysqlclient.impl.protocol.CommandType;
+import io.vertx.mysqlclient.typecodec.MySQLDataTypeCodecRegistry;
 import io.vertx.sqlclient.Tuple;
 import io.vertx.sqlclient.impl.command.CommandResponse;
 import io.vertx.sqlclient.impl.command.ExtendedQueryCommand;
@@ -42,7 +43,7 @@ void encode(MySQLEncoder encoder) {
     super.encode(encoder);
 
     if (statement.isCursorOpen) {
-      decoder = new RowResultDecoder<>(cmd.collector(), statement.rowDesc);
+      decoder = new RowResultDecoder<>(cmd.collector(), statement.rowDesc, encoder.socketConnection.getDataTypeCodecRegistry());
       sendStatementFetchCommand(statement.statementId, cmd.fetch());
     } else {
       Tuple params = cmd.params();
@@ -96,7 +97,7 @@ void decodePayload(ByteBuf payload, int payloadLength) {
             // need to reset packet number so that we can send a fetch request
             sequenceId = 0;
             // send fetch after cursor opened
-            decoder = new RowResultDecoder<>(cmd.collector(), statement.rowDesc);
+            decoder = new RowResultDecoder<>(cmd.collector(), statement.rowDesc, encoder.socketConnection.getDataTypeCodecRegistry());
 
             statement.isCursorOpen = true;
 
@@ -146,7 +147,7 @@ private void sendStatementExecuteCommand(MySQLPreparedStatement statement, boole
       for (int i = 0; i < numOfParams; i++) {
         Object value = params.getValue(i);
         if (value != null) {
-          DataTypeCodec.encodeBinary(statement.bindingTypes()[i], value, encoder.encodingCharset, packet);
+          encodeRowValue(packet, value, value.getClass());
         } else {
           nullBitmap[i / 8] |= (1 << (i & 7));
         }
@@ -163,6 +164,13 @@ private void sendStatementExecuteCommand(MySQLPreparedStatement statement, boole
     sendPacket(packet, payloadLength);
   }
 
+  @SuppressWarnings("unchecked")
+  private <T> void encodeRowValue(ByteBuf buf, T value, Class<?> clazz) {
+    MySQLDataTypeCodecRegistry dataTypeCodecRegistry = encoder.socketConnection.getDataTypeCodecRegistry();
+    io.vertx.sqlclient.codec.DataTypeCodec<T, ?> dataTypeCodec = (io.vertx.sqlclient.codec.DataTypeCodec<T, ?>) dataTypeCodecRegistry.lookupForEncoding(clazz);
+    dataTypeCodec.encode(buf, value);
+  }
+
   private void sendStatementFetchCommand(long statementId, int count) {
     ByteBuf packet = allocateBuffer();
     // encode packet header

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/QueryCommandBaseCodec.java

@@ -94,7 +94,7 @@ protected void handleResultsetColumnDefinitions(ByteBuf payload) {
 
   protected void handleResultsetColumnDefinitionsDecodingCompleted() {
     commandHandlerState = CommandHandlerState.HANDLING_ROW_DATA_OR_END_PACKET;
-    decoder = new RowResultDecoder<>(cmd.collector(), /*cmd.isSingleton()*/ new MySQLRowDesc(columnDefinitions, format));
+    decoder = new RowResultDecoder<>(cmd.collector(), /*cmd.isSingleton()*/ new MySQLRowDesc(columnDefinitions, format), encoder.socketConnection.getDataTypeCodecRegistry());
   }
 
   protected void handleRows(ByteBuf payload, int payloadLength) {