Refactor ByteBufCombiner#composite method to NettyBufferUtils#composite (#126)

Motivation:
Improve reusability by moving the `composite` method from the
`ByteBufCombiner` class to the `NettyBufferUtils` class.

Modifications:
The `composite` method, previously residing in the `ByteBufCombiner`
class, has been extracted and refactored into `composite` method within
the `NettyBufferUtils` class.

Result:
Enhance code organization.

Author: jchrys

src/main/java/io/asyncer/r2dbc/mysql/internal/util/NettyBufferUtils.java

@@ -17,17 +17,62 @@
 package io.asyncer.r2dbc.mysql.internal.util;
 
 import io.asyncer.r2dbc.mysql.message.FieldValue;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
 
 import java.util.List;
 
+
 /**
  * An internal utility considers the use of safe release buffers (array or {@link List}). It uses standard
  * netty {@link ReferenceCountUtil#safeRelease} to suppress release errors.
  */
 public final class NettyBufferUtils {
 
+    /**
+     * Combine {@link ByteBuf}s through composite buffer.
+     * <p>
+     * This method would release all {@link ByteBuf}s when any exception throws.
+     *
+     * @param parts The {@link ByteBuf}s want to be wrap, it can not be empty, and it will be cleared.
+     * @return A {@link ByteBuf} holds the all bytes of given {@code parts}, it may be a read-only buffer.
+     */
+    public static ByteBuf composite(final List<ByteBuf> parts) {
+        final int size = parts.size();
+
+        switch (size) {
+        case 0:
+            throw new IllegalStateException("No buffer available");
+        case 1:
+            try {
+                return parts.get(0);
+            } finally {
+                parts.clear();
+            }
+        default:
+            CompositeByteBuf composite = null;
+
+            try {
+                composite = parts.get(0).alloc().compositeBuffer(size);
+                // Auto-releasing failed parts
+                return composite.addComponents(true, parts);
+            } catch (Throwable e) {
+                if (composite == null) {
+                    // Alloc failed, release parts.
+                    releaseAll(parts);
+                } else {
+                    // Also release success parts.
+                    composite.release();
+                }
+                throw e;
+            } finally {
+                parts.clear();
+            }
+        }
+    }
+
     public static void releaseAll(ReferenceCounted[] parts) {
         for (ReferenceCounted counted : parts) {
             if (counted != null) {

src/main/java/io/asyncer/r2dbc/mysql/message/server/ByteBufCombiner.java

@@ -77,7 +77,7 @@ static FieldReader of(List<ByteBuf> buffers) {
 
         if (totalSize <= Integer.MAX_VALUE) {
             // The buffers will be cleared by ByteBufCombiner.composite().
-            ByteBuf combined = ByteBufCombiner.composite(buffers);
+            ByteBuf combined = NettyBufferUtils.composite(buffers);
             try {
                 return new NormalFieldReader(combined);
             } catch (Throwable e) {

src/main/java/io/asyncer/r2dbc/mysql/message/server/FieldReader.java

@@ -90,11 +90,12 @@ private static ServerMessage decodeMessage(List<ByteBuf> buffers, int envelopeId
         ConnectionContext context, DecodeContext decodeContext) {
         if (decodeContext instanceof ResultDecodeContext) {
             return decodeResult(buffers, context, (ResultDecodeContext) decodeContext);
-        } else if (decodeContext instanceof FetchDecodeContext) {
+        }
+        if (decodeContext instanceof FetchDecodeContext) {
             return decodeFetch(buffers, context);
         }
 
-        ByteBuf combined = ByteBufCombiner.composite(buffers);
+        ByteBuf combined = NettyBufferUtils.composite(buffers);
 
         try {
             if (decodeContext instanceof CommandDecodeContext) {
@@ -159,7 +160,7 @@ private static ServerMessage decodeResult(List<ByteBuf> buffers, ConnectionConte
         }
 
         if (decodeContext.isInMetadata()) {
-            ByteBuf combined = ByteBufCombiner.composite(buffers);
+            ByteBuf combined = NettyBufferUtils.composite(buffers);
             try {
                 return decodeInMetadata(combined, header, context, decodeContext);
             } finally {
@@ -309,7 +310,7 @@ private static ErrorMessage decodeCheckError(List<ByteBuf> buffers, short header
             // 0xFF is not header of var integer,
             // not header of text result null (0xFB) and
             // not header of column metadata (0x03 + "def")
-            ByteBuf combined = ByteBufCombiner.composite(buffers);
+            ByteBuf combined = NettyBufferUtils.composite(buffers);
             try {
                 return ErrorMessage.decode(combined);
             } finally {
@@ -329,15 +330,15 @@ private static ServerMessage decodeRow(List<ByteBuf> buffers, ByteBuf firstBuf,
             int byteSize = firstBuf.readableBytes();
 
             if (OkMessage.isValidSize(byteSize)) {
-                ByteBuf combined = ByteBufCombiner.composite(buffers);
+                ByteBuf combined = NettyBufferUtils.composite(buffers);
 
                 try {
                     return OkMessage.decode(combined, context);
                 } finally {
                     combined.release();
                 }
             } else if (EofMessage.isValidSize(byteSize)) {
-                ByteBuf combined = ByteBufCombiner.composite(buffers);
+                ByteBuf combined = NettyBufferUtils.composite(buffers);
 
                 try {
                     return EofMessage.decode(combined);