modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java - ignite-3 - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.ignite.client.handler;

 import java.io.IOException;
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;

 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import org.apache.ignite.app.Ignite;
 import org.apache.ignite.client.proto.ClientDataType;
 import org.apache.ignite.client.proto.ClientErrorCode;
 import org.apache.ignite.client.proto.ClientMessagePacker;
 import org.apache.ignite.client.proto.ClientMessageUnpacker;
 import org.apache.ignite.client.proto.ClientOp;
 import org.apache.ignite.client.proto.ProtocolVersion;
 import org.apache.ignite.client.proto.ServerMessageType;
 import org.apache.ignite.internal.schema.Column;
 import org.apache.ignite.internal.schema.NativeTypeSpec;
 import org.apache.ignite.internal.schema.SchemaAware;
 import org.apache.ignite.internal.schema.SchemaDescriptor;
 import org.apache.ignite.internal.table.IgniteTablesInternal;
 import org.apache.ignite.internal.table.TableImpl;
 import org.apache.ignite.internal.table.TupleBuilderImpl;
 import org.apache.ignite.lang.IgniteException;
 import org.apache.ignite.table.Table;
 import org.apache.ignite.table.Tuple;
 import org.msgpack.core.MessageFormat;
 import org.msgpack.core.buffer.ByteBufferInput;
 import org.slf4j.Logger;

 /**
  * Handles messages from thin clients.
  */
 @SuppressWarnings({"rawtypes", "unchecked"})
 public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter {
     /** Logger. */
     private final Logger log;

     /** API entry point. */
     private final Ignite ignite;

     /** Context. */
     private ClientContext clientContext;

     /**
      * Constructor.
      *
      * @param ignite Ignite API entry point.
      * @param log Logger.
      */
     public ClientInboundMessageHandler(Ignite ignite, Logger log) {
         assert ignite != null;
         assert log != null;

         this.ignite = ignite;
         this.log = log;
     }

     /** {@inheritDoc} */
     @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException {
         var buf = (ByteBuffer) msg;

         var unpacker = getUnpacker(buf);
         var packer = getPacker();

         if (clientContext == null)
             handshake(ctx, unpacker, packer);
         else
             processOperation(ctx, unpacker, packer);
     }

     private void handshake(ChannelHandlerContext ctx, ClientMessageUnpacker unpacker, ClientMessagePacker packer)
             throws IOException {
         try {
             var clientVer = ProtocolVersion.unpack(unpacker);

             if (!clientVer.equals(ProtocolVersion.LATEST_VER))
                 throw new IgniteException("Unsupported version: " +
                         clientVer.major() + "." + clientVer.minor() + "." + clientVer.patch());

             var clientCode = unpacker.unpackInt();
             var featuresLen = unpacker.unpackBinaryHeader();
             var features = BitSet.valueOf(unpacker.readPayload(featuresLen));

             clientContext = new ClientContext(clientVer, clientCode, features);

             log.debug("Handshake: " + clientContext);

             var extensionsLen = unpacker.unpackMapHeader();
             unpacker.skipValue(extensionsLen);

             // Response.
             ProtocolVersion.LATEST_VER.pack(packer);

             packer.packInt(ClientErrorCode.SUCCESS)
                     .packBinaryHeader(0) // Features.
                     .packMapHeader(0); // Extensions.

             write(packer, ctx);
         }
         catch (Throwable t) {
             packer = getPacker();

             ProtocolVersion.LATEST_VER.pack(packer);
             packer.packInt(ClientErrorCode.FAILED).packString(t.getMessage());

             write(packer, ctx);
         }
     }

     private void write(ClientMessagePacker packer, ChannelHandlerContext ctx) {
         var buf = packer.toMessageBuffer().sliceAsByteBuffer();

         ctx.writeAndFlush(buf);
     }

     private void writeError(int requestId, Throwable err, ChannelHandlerContext ctx) {
         try {
             assert err != null;

             ClientMessagePacker packer = getPacker();
             packer.packInt(ServerMessageType.RESPONSE);
             packer.packInt(requestId);
             packer.packInt(ClientErrorCode.FAILED);

             String msg = err.getMessage();

             if (msg == null)
                 msg = err.getClass().getName();

             packer.packString(msg);

             write(packer, ctx);
         }
         catch (Throwable t) {
             exceptionCaught(ctx, t);
         }
     }

     private ClientMessagePacker getPacker() {
         return new ClientMessagePacker();
     }

     private ClientMessageUnpacker getUnpacker(ByteBuffer buf) {
         return new ClientMessageUnpacker(new ByteBufferInput(buf));
     }

     private void processOperation(ChannelHandlerContext ctx, ClientMessageUnpacker unpacker, ClientMessagePacker packer) throws IOException {
         var opCode = unpacker.unpackInt();
         var requestId = unpacker.unpackInt();

         packer.packInt(ServerMessageType.RESPONSE)
                 .packInt(requestId)
                 .packInt(ClientErrorCode.SUCCESS);

         try {
             var fut = processOperation(unpacker, packer, opCode);

             if (fut == null) {
                 // Operation completed synchronously.
                 write(packer, ctx);
             }
             else {
                 fut.whenComplete((Object res, Object err) -> {
                     if (err != null)
                         writeError(requestId, (Throwable) err, ctx);
                     else
                         write(packer, ctx);
                 });
             }

         }
         catch (Throwable t) {
             writeError(requestId, t, ctx);
         }
     }

     private CompletableFuture processOperation(ClientMessageUnpacker unpacker, ClientMessagePacker packer, int opCode)
             throws IOException {
         // TODO: Handle all operations asynchronously (add async table API).
         switch (opCode) {
             case ClientOp.TABLE_DROP: {
                 var tableName = unpacker.unpackString();

                 ignite.tables().dropTable(tableName);

                 break;
             }

             case ClientOp.TABLES_GET: {
                 List<Table> tables = ignite.tables().tables();

                 packer.packMapHeader(tables.size());

                 for (var table : tables) {
                     var tableImpl = (TableImpl) table;

                     packer.packUuid(tableImpl.tableId());
                     packer.packString(table.tableName());
                 }

                 break;
             }

             case ClientOp.SCHEMAS_GET: {
                 var table = readTable(unpacker);

                 if (unpacker.getNextFormat() == MessageFormat.NIL) {
                     // Return the latest schema.
                     packer.packMapHeader(1);

                     var schema = table.schemaView().schema();

                     if (schema == null)
                         throw new IgniteException("Schema registry is not initialized.");

                     writeSchema(packer, schema.version(), schema);
                 }
                 else {
                     var cnt = unpacker.unpackArrayHeader();
                     packer.packMapHeader(cnt);

                     for (var i = 0; i < cnt; i++) {
                         var schemaVer = unpacker.unpackInt();
                         var schema = table.schemaView().schema(schemaVer);
                         writeSchema(packer, schemaVer, schema);
                     }
                 }

                 break;
             }

             case ClientOp.TABLE_GET: {
                 String tableName = unpacker.unpackString();
                 Table table = ignite.tables().table(tableName);

                 if (table == null)
                     packer.packNil();
                 else
                     packer.packUuid(((TableImpl) table).tableId());

                 break;
             }

             case ClientOp.TUPLE_UPSERT: {
                 var table = readTable(unpacker);
                 var tuple = readTuple(unpacker, table, false);

                 return table.upsertAsync(tuple);
             }

             case ClientOp.TUPLE_UPSERT_SCHEMALESS: {
                 var table = readTable(unpacker);
                 var tuple = readTupleSchemaless(unpacker, table);

                 return table.upsertAsync(tuple);
             }

             case ClientOp.TUPLE_GET: {
                 var table = readTable(unpacker);
                 var keyTuple = readTuple(unpacker, table, true);

                 return table.getAsync(keyTuple).thenAccept(t -> writeTuple(packer, t));
             }

             default:
                 throw new IgniteException("Unexpected operation code: " + opCode);
         }

         return null;
     }

     private void writeSchema(ClientMessagePacker packer, int schemaVer, SchemaDescriptor schema) throws IOException {
         packer.packInt(schemaVer);

         if (schema == null) {
             packer.packNil();

             return;
         }

         var colCnt = schema.columnNames().size();
         packer.packArrayHeader(colCnt);

         for (var colIdx = 0; colIdx < colCnt; colIdx++) {
             var col = schema.column(colIdx);

             packer.packArrayHeader(4);
             packer.packString(col.name());
             packer.packInt(getClientDataType(col.type().spec()));
             packer.packBoolean(schema.isKeyColumn(colIdx));
             packer.packBoolean(col.nullable());
         }
     }

     private void writeTuple(ClientMessagePacker packer, Tuple tuple) {
         try {
             if (tuple == null) {
                 packer.packNil();

                 return;
             }

             var schema = ((SchemaAware) tuple).schema();

             packer.packInt(schema.version());

             for (var col : schema.keyColumns().columns())
                 writeColumnValue(packer, tuple, col);

             for (var col : schema.valueColumns().columns())
                 writeColumnValue(packer, tuple, col);
         }
         catch (Throwable t) {
             throw new IgniteException("Failed to serialize tuple", t);
         }
     }

     private Tuple readTuple(ClientMessageUnpacker unpacker, TableImpl table, boolean keyOnly) throws IOException {
         var schemaId = unpacker.unpackInt();
         var schema = table.schemaView().schema(schemaId);
         var builder = (TupleBuilderImpl) table.tupleBuilder();

         var cnt = keyOnly ? schema.keyColumns().length() : schema.length();

         for (int i = 0; i < cnt; i++) {
             if (unpacker.getNextFormat() == MessageFormat.NIL) {
                 unpacker.skipValue();
                 continue;
             }

             readAndSetColumnValue(unpacker, builder, schema.column(i));
         }

         return builder.build();
     }

     private Tuple readTupleSchemaless(ClientMessageUnpacker unpacker, TableImpl table) throws IOException {
         var cnt = unpacker.unpackMapHeader();
         var builder = table.tupleBuilder();

         for (int i = 0; i < cnt; i++) {
             var colName = unpacker.unpackString();

             builder.set(colName, unpacker.unpackValue());
         }

         return builder.build();
     }

     private TableImpl readTable(ClientMessageUnpacker unpacker) throws IOException {
         var tableId = unpacker.unpackUuid();

         return ((IgniteTablesInternal)ignite.tables()).table(tableId);
     }

     private void readAndSetColumnValue(ClientMessageUnpacker unpacker, TupleBuilderImpl builder, Column col)
             throws IOException {
         builder.set(col.name(), unpacker.unpackObject(getClientDataType(col.type().spec())));
     }

     private static int getClientDataType(NativeTypeSpec spec) {
         switch (spec) {
             case INT8:
                 return ClientDataType.INT8;

             case INT16:
                 return ClientDataType.INT16;

             case INT32:
                 return ClientDataType.INT32;

             case INT64:
                 return ClientDataType.INT64;

             case FLOAT:
                 return ClientDataType.FLOAT;

             case DOUBLE:
                 return ClientDataType.DOUBLE;

             case DECIMAL:
                 return ClientDataType.DECIMAL;

             case UUID:
                 return ClientDataType.UUID;

             case STRING:
                 return ClientDataType.STRING;

             case BYTES:
                 return ClientDataType.BYTES;

             case BITMASK:
                 return ClientDataType.BITMASK;
         }

         throw new IgniteException("Unsupported native type: " + spec);
     }

     private void writeColumnValue(ClientMessagePacker packer, Tuple tuple, Column col) throws IOException {
         var val = tuple.value(col.name());

         if (val == null) {
             packer.packNil();
             return;
         }

         switch (col.type().spec()) {
             case INT8:
                 packer.packByte((byte) val);
                 break;

             case INT16:
                 packer.packShort((short) val);
                 break;

             case INT32:
                 packer.packInt((int) val);
                 break;

             case INT64:
                 packer.packLong((long) val);
                 break;

             case FLOAT:
                 packer.packFloat((float) val);
                 break;

             case DOUBLE:
                 packer.packDouble((double) val);
                 break;

             case DECIMAL:
                 packer.packDecimal((BigDecimal) val);
                 break;

             case UUID:
                 packer.packUuid((UUID) val);
                 break;

             case STRING:
                 packer.packString((String) val);
                 break;

             case BYTES:
                 byte[] bytes = (byte[]) val;
                 packer.packBinaryHeader(bytes.length);
                 packer.writePayload(bytes);
                 break;

             case BITMASK:
                 packer.packBitSet((BitSet)val);
                 break;

             default:
                 throw new IgniteException("Data type not supported: " + col.type());
         }
     }

     /** {@inheritDoc} */
     @Override public void channelReadComplete(ChannelHandlerContext ctx) {
         ctx.flush();
     }

     /** {@inheritDoc} */
     @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         log.error(cause.getMessage(), cause);

         ctx.close();
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.ignite.client.handler;

	import java.io.IOException;
	import java.math.BigDecimal;
	import java.nio.ByteBuffer;
	import java.util.BitSet;
	import java.util.List;
	import java.util.UUID;
	import java.util.concurrent.CompletableFuture;

	import io.netty.channel.ChannelHandlerContext;
	import io.netty.channel.ChannelInboundHandlerAdapter;
	import org.apache.ignite.app.Ignite;
	import org.apache.ignite.client.proto.ClientDataType;
	import org.apache.ignite.client.proto.ClientErrorCode;
	import org.apache.ignite.client.proto.ClientMessagePacker;
	import org.apache.ignite.client.proto.ClientMessageUnpacker;
	import org.apache.ignite.client.proto.ClientOp;
	import org.apache.ignite.client.proto.ProtocolVersion;
	import org.apache.ignite.client.proto.ServerMessageType;
	import org.apache.ignite.internal.schema.Column;
	import org.apache.ignite.internal.schema.NativeTypeSpec;
	import org.apache.ignite.internal.schema.SchemaAware;
	import org.apache.ignite.internal.schema.SchemaDescriptor;
	import org.apache.ignite.internal.table.IgniteTablesInternal;
	import org.apache.ignite.internal.table.TableImpl;
	import org.apache.ignite.internal.table.TupleBuilderImpl;
	import org.apache.ignite.lang.IgniteException;
	import org.apache.ignite.table.Table;
	import org.apache.ignite.table.Tuple;
	import org.msgpack.core.MessageFormat;
	import org.msgpack.core.buffer.ByteBufferInput;
	import org.slf4j.Logger;

	/**
	* Handles messages from thin clients.
	*/
	@SuppressWarnings({"rawtypes", "unchecked"})
	public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter {
	/** Logger. */
	private final Logger log;

	/** API entry point. */
	private final Ignite ignite;

	/** Context. */
	private ClientContext clientContext;

	/**
	* Constructor.
	*
	* @param ignite Ignite API entry point.
	* @param log Logger.
	*/
	public ClientInboundMessageHandler(Ignite ignite, Logger log) {
	assert ignite != null;
	assert log != null;

	this.ignite = ignite;
	this.log = log;
	}

	/** {@inheritDoc} */
	@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException {
	var buf = (ByteBuffer) msg;

	var unpacker = getUnpacker(buf);
	var packer = getPacker();

	if (clientContext == null)
	handshake(ctx, unpacker, packer);
	else
	processOperation(ctx, unpacker, packer);
	}

	private void handshake(ChannelHandlerContext ctx, ClientMessageUnpacker unpacker, ClientMessagePacker packer)
	throws IOException {
	try {
	var clientVer = ProtocolVersion.unpack(unpacker);

	if (!clientVer.equals(ProtocolVersion.LATEST_VER))
	throw new IgniteException("Unsupported version: " +
	clientVer.major() + "." + clientVer.minor() + "." + clientVer.patch());

	var clientCode = unpacker.unpackInt();
	var featuresLen = unpacker.unpackBinaryHeader();
	var features = BitSet.valueOf(unpacker.readPayload(featuresLen));

	clientContext = new ClientContext(clientVer, clientCode, features);

	log.debug("Handshake: " + clientContext);

	var extensionsLen = unpacker.unpackMapHeader();
	unpacker.skipValue(extensionsLen);

	// Response.
	ProtocolVersion.LATEST_VER.pack(packer);

	packer.packInt(ClientErrorCode.SUCCESS)
	.packBinaryHeader(0) // Features.
	.packMapHeader(0); // Extensions.

	write(packer, ctx);
	}
	catch (Throwable t) {
	packer = getPacker();

	ProtocolVersion.LATEST_VER.pack(packer);
	packer.packInt(ClientErrorCode.FAILED).packString(t.getMessage());

	write(packer, ctx);
	}
	}

	private void write(ClientMessagePacker packer, ChannelHandlerContext ctx) {
	var buf = packer.toMessageBuffer().sliceAsByteBuffer();

	ctx.writeAndFlush(buf);
	}

	private void writeError(int requestId, Throwable err, ChannelHandlerContext ctx) {
	try {
	assert err != null;

	ClientMessagePacker packer = getPacker();
	packer.packInt(ServerMessageType.RESPONSE);
	packer.packInt(requestId);
	packer.packInt(ClientErrorCode.FAILED);

	String msg = err.getMessage();

	if (msg == null)
	msg = err.getClass().getName();

	packer.packString(msg);

	write(packer, ctx);
	}
	catch (Throwable t) {
	exceptionCaught(ctx, t);
	}
	}

	private ClientMessagePacker getPacker() {
	return new ClientMessagePacker();
	}

	private ClientMessageUnpacker getUnpacker(ByteBuffer buf) {
	return new ClientMessageUnpacker(new ByteBufferInput(buf));
	}

	private void processOperation(ChannelHandlerContext ctx, ClientMessageUnpacker unpacker, ClientMessagePacker packer) throws IOException {
	var opCode = unpacker.unpackInt();
	var requestId = unpacker.unpackInt();

	packer.packInt(ServerMessageType.RESPONSE)
	.packInt(requestId)
	.packInt(ClientErrorCode.SUCCESS);

	try {
	var fut = processOperation(unpacker, packer, opCode);

	if (fut == null) {
	// Operation completed synchronously.
	write(packer, ctx);
	}
	else {
	fut.whenComplete((Object res, Object err) -> {
	if (err != null)
	writeError(requestId, (Throwable) err, ctx);
	else
	write(packer, ctx);
	});
	}

	}
	catch (Throwable t) {
	writeError(requestId, t, ctx);
	}
	}

	private CompletableFuture processOperation(ClientMessageUnpacker unpacker, ClientMessagePacker packer, int opCode)
	throws IOException {
	// TODO: Handle all operations asynchronously (add async table API).
	switch (opCode) {
	case ClientOp.TABLE_DROP: {
	var tableName = unpacker.unpackString();

	ignite.tables().dropTable(tableName);

	break;
	}

	case ClientOp.TABLES_GET: {
	List<Table> tables = ignite.tables().tables();

	packer.packMapHeader(tables.size());

	for (var table : tables) {
	var tableImpl = (TableImpl) table;

	packer.packUuid(tableImpl.tableId());
	packer.packString(table.tableName());
	}

	break;
	}

	case ClientOp.SCHEMAS_GET: {
	var table = readTable(unpacker);

	if (unpacker.getNextFormat() == MessageFormat.NIL) {
	// Return the latest schema.
	packer.packMapHeader(1);

	var schema = table.schemaView().schema();

	if (schema == null)
	throw new IgniteException("Schema registry is not initialized.");

	writeSchema(packer, schema.version(), schema);
	}
	else {
	var cnt = unpacker.unpackArrayHeader();
	packer.packMapHeader(cnt);

	for (var i = 0; i < cnt; i++) {
	var schemaVer = unpacker.unpackInt();
	var schema = table.schemaView().schema(schemaVer);
	writeSchema(packer, schemaVer, schema);
	}
	}

	break;
	}

	case ClientOp.TABLE_GET: {
	String tableName = unpacker.unpackString();
	Table table = ignite.tables().table(tableName);

	if (table == null)
	packer.packNil();
	else
	packer.packUuid(((TableImpl) table).tableId());

	break;
	}

	case ClientOp.TUPLE_UPSERT: {
	var table = readTable(unpacker);
	var tuple = readTuple(unpacker, table, false);

	return table.upsertAsync(tuple);
	}

	case ClientOp.TUPLE_UPSERT_SCHEMALESS: {
	var table = readTable(unpacker);
	var tuple = readTupleSchemaless(unpacker, table);

	return table.upsertAsync(tuple);
	}

	case ClientOp.TUPLE_GET: {
	var table = readTable(unpacker);
	var keyTuple = readTuple(unpacker, table, true);

	return table.getAsync(keyTuple).thenAccept(t -> writeTuple(packer, t));
	}

	default:
	throw new IgniteException("Unexpected operation code: " + opCode);
	}

	return null;
	}

	private void writeSchema(ClientMessagePacker packer, int schemaVer, SchemaDescriptor schema) throws IOException {
	packer.packInt(schemaVer);

	if (schema == null) {
	packer.packNil();

	return;
	}

	var colCnt = schema.columnNames().size();
	packer.packArrayHeader(colCnt);

	for (var colIdx = 0; colIdx < colCnt; colIdx++) {
	var col = schema.column(colIdx);

	packer.packArrayHeader(4);
	packer.packString(col.name());
	packer.packInt(getClientDataType(col.type().spec()));
	packer.packBoolean(schema.isKeyColumn(colIdx));
	packer.packBoolean(col.nullable());
	}
	}

	private void writeTuple(ClientMessagePacker packer, Tuple tuple) {
	try {
	if (tuple == null) {
	packer.packNil();

	return;
	}

	var schema = ((SchemaAware) tuple).schema();

	packer.packInt(schema.version());

	for (var col : schema.keyColumns().columns())
	writeColumnValue(packer, tuple, col);

	for (var col : schema.valueColumns().columns())
	writeColumnValue(packer, tuple, col);
	}
	catch (Throwable t) {
	throw new IgniteException("Failed to serialize tuple", t);
	}
	}

	private Tuple readTuple(ClientMessageUnpacker unpacker, TableImpl table, boolean keyOnly) throws IOException {
	var schemaId = unpacker.unpackInt();
	var schema = table.schemaView().schema(schemaId);
	var builder = (TupleBuilderImpl) table.tupleBuilder();

	var cnt = keyOnly ? schema.keyColumns().length() : schema.length();

	for (int i = 0; i < cnt; i++) {
	if (unpacker.getNextFormat() == MessageFormat.NIL) {
	unpacker.skipValue();
	continue;
	}

	readAndSetColumnValue(unpacker, builder, schema.column(i));
	}

	return builder.build();
	}

	private Tuple readTupleSchemaless(ClientMessageUnpacker unpacker, TableImpl table) throws IOException {
	var cnt = unpacker.unpackMapHeader();
	var builder = table.tupleBuilder();

	for (int i = 0; i < cnt; i++) {
	var colName = unpacker.unpackString();

	builder.set(colName, unpacker.unpackValue());
	}

	return builder.build();
	}

	private TableImpl readTable(ClientMessageUnpacker unpacker) throws IOException {
	var tableId = unpacker.unpackUuid();

	return ((IgniteTablesInternal)ignite.tables()).table(tableId);
	}

	private void readAndSetColumnValue(ClientMessageUnpacker unpacker, TupleBuilderImpl builder, Column col)
	throws IOException {
	builder.set(col.name(), unpacker.unpackObject(getClientDataType(col.type().spec())));
	}

	private static int getClientDataType(NativeTypeSpec spec) {
	switch (spec) {
	case INT8:
	return ClientDataType.INT8;

	case INT16:
	return ClientDataType.INT16;

	case INT32:
	return ClientDataType.INT32;

	case INT64:
	return ClientDataType.INT64;

	case FLOAT:
	return ClientDataType.FLOAT;

	case DOUBLE:
	return ClientDataType.DOUBLE;

	case DECIMAL:
	return ClientDataType.DECIMAL;

	case UUID:
	return ClientDataType.UUID;

	case STRING:
	return ClientDataType.STRING;

	case BYTES:
	return ClientDataType.BYTES;

	case BITMASK:
	return ClientDataType.BITMASK;
	}

	throw new IgniteException("Unsupported native type: " + spec);
	}

	private void writeColumnValue(ClientMessagePacker packer, Tuple tuple, Column col) throws IOException {
	var val = tuple.value(col.name());

	if (val == null) {
	packer.packNil();
	return;
	}

	switch (col.type().spec()) {
	case INT8:
	packer.packByte((byte) val);
	break;

	case INT16:
	packer.packShort((short) val);
	break;

	case INT32:
	packer.packInt((int) val);
	break;

	case INT64:
	packer.packLong((long) val);
	break;

	case FLOAT:
	packer.packFloat((float) val);
	break;

	case DOUBLE:
	packer.packDouble((double) val);
	break;

	case DECIMAL:
	packer.packDecimal((BigDecimal) val);
	break;

	case UUID:
	packer.packUuid((UUID) val);
	break;

	case STRING:
	packer.packString((String) val);
	break;

	case BYTES:
	byte[] bytes = (byte[]) val;
	packer.packBinaryHeader(bytes.length);
	packer.writePayload(bytes);
	break;

	case BITMASK:
	packer.packBitSet((BitSet)val);
	break;

	default:
	throw new IgniteException("Data type not supported: " + col.type());
	}
	}

	/** {@inheritDoc} */
	@Override public void channelReadComplete(ChannelHandlerContext ctx) {
	ctx.flush();
	}

	/** {@inheritDoc} */
	@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
	log.error(cause.getMessage(), cause);

	ctx.close();
	}
	}