Correct unsigned long handling (#161)

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

@@ -47,6 +47,17 @@ public final class VarIntUtils {
 
     private static final int MEDIUM_SIZE = MEDIUM_BYTES * Byte.SIZE;
 
+    /**
+     * Reads a length encoded integer from the given buffers. Notice that a length encoded integer can be
+     * greater than {@link Long#MAX_VALUE}. In this case it should be used as an unsigned long. If we need
+     * assume the result as a smaller integer, add code comment to explain it.
+     * <p>
+     * Note: it will change {@code firstPart} and {@code secondPart} readerIndex if necessary.
+     *
+     * @param firstPart the first part of a readable buffer include a part of the var integer.
+     * @param secondPart the second part of a readable buffer include subsequent part of the var integer.
+     * @return A var integer read from buffer.
+     */
     public static long crossReadVarInt(ByteBuf firstPart, ByteBuf secondPart) {
         requireNonNull(firstPart, "firstPart must not be null");
         requireNonNull(secondPart, "secondPart must not be null");
@@ -87,6 +98,10 @@ public static long crossReadVarInt(ByteBuf firstPart, ByteBuf secondPart) {
     }
 
     /**
+     * Reads a length encoded integer from the given buffer. Notice that a length encoded integer can be
+     * greater than {@link Long#MAX_VALUE}. In this case it should be used as an unsigned long. If we need
+     * assume the result as a smaller integer, add code comment to explain it.
+     * <p>
      * Note: it will change {@code buf} readerIndex.
      *
      * @param buf a readable buffer include a var integer.

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

@@ -40,6 +40,7 @@ public int getTotalColumns() {
     }
 
     static ColumnCountMessage decode(ByteBuf buf) {
+        // JVM does NOT support arrays longer than Integer.MAX_VALUE
         return new ColumnCountMessage(Math.toIntExact(VarIntUtils.readVarInt(buf)));
     }
 

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

@@ -173,7 +173,8 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         String column = readVarIntSizedString(buf, charset);
         String originColumn = readVarIntSizedString(buf, charset);
 
-        VarIntUtils.readVarInt(buf); // skip constant 0x0c encoded by var integer
+        // Skip constant 0x0c encoded by var integer
+        VarIntUtils.readVarInt(buf);
 
         int collationId = buf.readUnsignedShortLE();
         long size = buf.readUnsignedIntLE();
@@ -185,7 +186,7 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
     }
 
     private static String readVarIntSizedString(ByteBuf buf, Charset charset) {
-        // JVM can NOT support string which length upper than maximum of int32
+        // JVM does NOT support strings longer than Integer.MAX_VALUE
         int bytes = (int) VarIntUtils.readVarInt(buf);
 
         if (bytes == 0) {

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

@@ -99,12 +99,12 @@ public FieldValue readVarIntSizedField() {
             fieldSize = VarIntUtils.readVarInt(currentBuf);
         }
 
-        // Refresh non empty buffer because current buffer has been read.
+        // Refresh non-empty buffer because current buffer has been read.
         currentBuf = nonEmptyBuffer();
 
         List<ByteBuf> results = readSlice(currentBuf, fieldSize);
 
-        if (fieldSize > Integer.MAX_VALUE) {
+        if (Long.compareUnsigned(fieldSize, Integer.MAX_VALUE) > 0) {
             return retainedLargeField(results);
         }
 
@@ -130,26 +130,30 @@ protected void deallocate() {
      * list instead of a single buffer.
      *
      * @param current the current {@link ByteBuf} in {@link #buffers}.
-     * @param length  the length of read.
+     * @param length  the length of read, it can be an unsigned long.
      * @return result buffer list, should NEVER retain any buffer.
      */
     private List<ByteBuf> readSlice(ByteBuf current, long length) {
         ByteBuf buf = current;
         List<ByteBuf> results = new ArrayList<>(Math.max(
-            (int) Math.min((length / Envelopes.MAX_ENVELOPE_SIZE) + 2, Byte.MAX_VALUE), 10));
+            (int) Math.min(Long.divideUnsigned(length, Envelopes.MAX_ENVELOPE_SIZE) + 2, Byte.MAX_VALUE),
+            10
+        ));
         long totalSize = 0;
         int bufReadable;
 
         // totalSize + bufReadable <= length
-        while (totalSize <= length - (bufReadable = buf.readableBytes())) {
+        while (Long.compareUnsigned(totalSize, length - (bufReadable = buf.readableBytes())) <= 0) {
             totalSize += bufReadable;
             // No need readSlice because currentBufIndex will be increment after List pushed.
             results.add(buf);
             buf = this.buffers[++this.currentBufIndex];
         }
 
-        if (length > totalSize) {
-            // need bytes = length - `results` real length = length - (totalSize - `buf` length)
+        // totalSize < length
+        if (Long.compareUnsigned(length, totalSize) > 0) {
+            // need bytes = length - `results` length = length - totalSize
+            // length - totalSize should be an int due to while loop above.
             results.add(buf.readSlice((int) (length - totalSize)));
         } // else results has filled by prev buffer, and currentBufIndex is unread for now.
 

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

@@ -105,6 +105,7 @@ public boolean release(int decrement) {
     }
 
     private static ByteBuf readVarIntSizedRetained(ByteBuf buf) {
+        // Normal field will NEVER be greater than Integer.MAX_VALUE.
         int size = (int) VarIntUtils.readVarInt(buf);
         if (size == 0) {
             // Use EmptyByteBuf, new buffer no need to be retained.

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

@@ -160,10 +160,11 @@ static OkMessage decode(ByteBuf buf, ConnectionContext context) {
             if (size > sizeAfterVarInt) {
                 information = buf.toString(readerIndex, buf.writerIndex() - readerIndex, charset);
             } else {
+                // JVM does NOT support strings longer than Integer.MAX_VALUE
                 information = buf.toString(buf.readerIndex(), (int) size, charset);
             }
 
-            // Ignore session track, it is not human readable and useless for R2DBC client.
+            // Ignore session track, it is not human-readable and useless for R2DBC client.
             return new OkMessage(affectedRows, lastInsertId, serverStatuses, warnings, information);
         }