Implement basic BATCH in Firebird Proxy

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/codec/FirebirdPacketCodecEngine.java

@@ -54,18 +54,26 @@ public boolean isValidHeader(final int readableBytes) {
 
     @Override
     public void decode(final ChannelHandlerContext context, final ByteBuf in, final List<Object> out) {
-        if (pendingMessages.isEmpty()) {
-            int type = in.getInt(in.readerIndex());
-            pendingPacketType = FirebirdCommandPacketType.valueOf(type);
-            if (pendingPacketType == FirebirdCommandPacketType.ALLOCATE_STATEMENT) {
-                handleMultiPacket(context, in, out, ALLOCATE_STATEMENT_REQUEST_PAYLOAD_LENGTH);
-                return;
-            } else if (pendingPacketType == FirebirdCommandPacketType.FREE_STATEMENT) {
-                handleMultiPacket(context, in, out, FREE_STATEMENT_REQUEST_PAYLOAD_LENGTH);
-                return;
+        try {
+            if (pendingMessages.isEmpty()) {
+                int type = in.getInt(in.readerIndex());
+                pendingPacketType = FirebirdCommandPacketType.valueOf(type);
+                if (pendingPacketType == FirebirdCommandPacketType.ALLOCATE_STATEMENT) {
+                    handleMultiPacket(context, in, out, ALLOCATE_STATEMENT_REQUEST_PAYLOAD_LENGTH);
+                    return;
+                } else if (pendingPacketType == FirebirdCommandPacketType.FREE_STATEMENT) {
+                    handleMultiPacket(context, in, out, FREE_STATEMENT_REQUEST_PAYLOAD_LENGTH);
+                    return;
+                }
             }
+            addToBuffer(context, in, out);
+            // CHECKSTYLE:OFF
+        } catch (final RuntimeException ex) {
+            // CHECKSTYLE:ON
+            // TODO replace with a proper Firebird error response packet; for now close the channel so the client fails fast instead of hanging.
+            context.close();
+            throw ex;
         }
-        addToBuffer(context, in, out);
     }
 
     private void handleMultiPacket(final ChannelHandlerContext context, final ByteBuf in, final List<Object> out, final int firstPacketLength) {
@@ -106,7 +114,7 @@ private void processPackets(final ChannelHandlerContext context, final ByteBuf b
                 return;
             }
             int readerIndex = buffer.readerIndex();
-            FirebirdCommandPacketType commandType = FirebirdCommandPacketType.valueOf(buffer.getInt(readerIndex));
+            FirebirdCommandPacketType commandType = (pendingPacketType != null) ? pendingPacketType : FirebirdCommandPacketType.valueOf(buffer.getInt(readerIndex));
             if (FirebirdCommandPacketType.VOID == commandType) {
                 buffer.skipBytes(MESSAGE_TYPE_LENGTH);
                 continue;
@@ -129,11 +137,11 @@ private int findPacketLength(final ChannelHandlerContext context, final ByteBuf
         try {
             FirebirdPacketPayload payload = new FirebirdPacketPayload(slice, charset);
             int expectedLength = FirebirdCommandPacketFactory.getExpectedLength(commandType, payload,
-                    context.channel().attr(FirebirdConstant.CONNECTION_PROTOCOL_VERSION).get());
-            if (expectedLength <= 0) {
-                return readableBytes;
+                    context.channel().attr(FirebirdConstant.CONNECTION_PROTOCOL_VERSION).get(), context.channel().attr(FirebirdConstant.CURRENT_CONNECTION).get());
+            if (expectedLength < 0) {
+                return -1;
             }
-            return readableBytes >= expectedLength ? expectedLength : -1;
+            return 0 == expectedLength ? readableBytes : (readableBytes >= expectedLength ? expectedLength : -1);
         } catch (final IndexOutOfBoundsException ex) {
             return -1;
         } finally {
@@ -150,7 +158,9 @@ public void encode(final ChannelHandlerContext context, final DatabasePacket mes
         } catch (final RuntimeException ex) {
             // CHECKSTYLE:ON
             payload.getByteBuf().resetWriterIndex();
-            // TODO send error packet
+            // TODO replace with a proper Firebird error response packet; for now close the channel so the client fails fast instead of hanging.
+            context.close();
+            throw ex;
         }
     }
 

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/constant/FirebirdConstant.java

@@ -29,4 +29,6 @@
 public final class FirebirdConstant {
 
     public static final AttributeKey<FirebirdProtocolVersion> CONNECTION_PROTOCOL_VERSION = AttributeKey.valueOf("FIREBIRD_CONNECTION_PROTOCOL_VERSION");
+
+    public static final AttributeKey<Integer> CURRENT_CONNECTION = AttributeKey.valueOf("FIREBIRD_CURRENT_CONNECTION");
 }

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/FirebirdCommandPacketFactory.java

@@ -21,6 +21,11 @@
 import lombok.NoArgsConstructor;
 import org.apache.shardingsphere.database.protocol.firebird.constant.protocol.FirebirdProtocolVersion;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.admin.FirebirdUnsupportedCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchCancelCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchCreateCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchExecuteCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchReleaseCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchSendMessageCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdGetBlobSegmentCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdOpenBlobCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdPutBlobSegmentCommandPacket;
@@ -54,9 +59,11 @@ public final class FirebirdCommandPacketFactory {
      * @param commandPacketType command packet type for Firebird
      * @param payload packet payload for Firebird
      * @param protocolVersion protocol version of Firebird
+     * @param connectionId connection ID
      * @return created instance
      */
-    public static FirebirdCommandPacket newInstance(final FirebirdCommandPacketType commandPacketType, final FirebirdPacketPayload payload, final FirebirdProtocolVersion protocolVersion) {
+    public static FirebirdCommandPacket newInstance(final FirebirdCommandPacketType commandPacketType, final FirebirdPacketPayload payload, final FirebirdProtocolVersion protocolVersion,
+                                                    final int connectionId) {
         switch (commandPacketType) {
             case INFO_DATABASE:
                 return FirebirdDatabaseInfoPacketType.createPacket(payload);
@@ -97,6 +104,16 @@ public static FirebirdCommandPacket newInstance(final FirebirdCommandPacketType
                 return new FirebirdRollbackTransactionPacket(payload);
             case FREE_STATEMENT:
                 return new FirebirdFreeStatementPacket(payload);
+            case BATCH_CREATE:
+                return new FirebirdBatchCreateCommandPacket(payload);
+            case BATCH_MSG:
+                return new FirebirdBatchSendMessageCommandPacket(payload);
+            case BATCH_EXEC:
+                return new FirebirdBatchExecuteCommandPacket(payload);
+            case BATCH_RLS:
+                return new FirebirdBatchReleaseCommandPacket(payload);
+            case BATCH_CANCEL:
+                return new FirebirdBatchCancelCommandPacket(payload);
             default:
                 return new FirebirdUnsupportedCommandPacket(commandPacketType);
         }
@@ -108,14 +125,15 @@ public static FirebirdCommandPacket newInstance(final FirebirdCommandPacketType
      * @param commandPacketType command packet type for Firebird
      * @param payload packet payload for Firebird
      * @param protocolVersion protocol version of Firebird
+     * @param connectionId connection ID
      * @return expected length of packet, or 0 if length is variable
      */
-    public static int getExpectedLength(final FirebirdCommandPacketType commandPacketType, final FirebirdPacketPayload payload, final FirebirdProtocolVersion protocolVersion) {
-        return getLength(commandPacketType, payload, protocolVersion);
+    public static int getExpectedLength(final FirebirdCommandPacketType commandPacketType, final FirebirdPacketPayload payload, final FirebirdProtocolVersion protocolVersion, final int connectionId) {
+        return getLength(commandPacketType, payload, protocolVersion, connectionId);
     }
 
     private static int getLength(final FirebirdCommandPacketType commandPacketType, final FirebirdPacketPayload payload,
-                                 final FirebirdProtocolVersion protocolVersion) throws IndexOutOfBoundsException {
+                                 final FirebirdProtocolVersion protocolVersion, final int connectionId) throws IndexOutOfBoundsException {
         switch (commandPacketType) {
             case INFO_DATABASE:
             case INFO_SQL:
@@ -154,6 +172,16 @@ private static int getLength(final FirebirdCommandPacketType commandPacketType,
                 return FirebirdRollbackTransactionPacket.getLength();
             case FREE_STATEMENT:
                 return FirebirdFreeStatementPacket.getLength();
+            case BATCH_CREATE:
+                return FirebirdBatchCreateCommandPacket.getLength(payload);
+            case BATCH_MSG:
+                return FirebirdBatchSendMessageCommandPacket.getLength(payload, connectionId);
+            case BATCH_EXEC:
+                return FirebirdBatchExecuteCommandPacket.getLength();
+            case BATCH_RLS:
+                return FirebirdBatchReleaseCommandPacket.getLength();
+            case BATCH_CANCEL:
+                return FirebirdBatchCancelCommandPacket.getLength();
             default:
                 return 0;
         }

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchCancelCommandPacket.java

@@ -0,0 +1,49 @@
+/*
+ * 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.shardingsphere.database.protocol.firebird.packet.command.query.batch;
+
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+/**
+ * Firebird batch cancel command packet.
+ */
+@Getter
+public final class FirebirdBatchCancelCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    public FirebirdBatchCancelCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @return length of packet
+     */
+    public static int getLength() {
+        return 8;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchCreateCommandPacket.java

@@ -0,0 +1,68 @@
+/*
+ * 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.shardingsphere.database.protocol.firebird.packet.command.query.batch;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.exception.FirebirdProtocolException;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+/**
+ * Firebird create batch command packet.
+ */
+@Getter
+public final class FirebirdBatchCreateCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    private final ByteBuf batchBlr;
+
+    private final long batchMessageLength;
+
+    private final ByteBuf batchParametersBuffer;
+
+    public FirebirdBatchCreateCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+        batchBlr = payload.readBuffer();
+        if (!batchBlr.isReadable()) {
+            throw new FirebirdProtocolException("Missing required format info in createBatch()");
+        }
+        batchMessageLength = payload.readInt4Unsigned();
+        batchParametersBuffer = payload.readBuffer();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @param payload Firebird packet payload
+     * @return length of packet
+     */
+    public static int getLength(final FirebirdPacketPayload payload) {
+        int length = 8;
+        length += payload.getBufferLength(length);
+        length += 4;
+        length += payload.getBufferLength(length);
+        return length;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchExecuteCommandPacket.java

@@ -0,0 +1,49 @@
+/*
+ * 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.shardingsphere.database.protocol.firebird.packet.command.query.batch;
+
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+@Getter
+public class FirebirdBatchExecuteCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    private final int transactionHandle;
+
+    public FirebirdBatchExecuteCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+        transactionHandle = payload.readInt4();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @return length of packet
+     */
+    public static int getLength() {
+        return 12;
+    }
+}