Code review changes

dfsclient/src/main/java/org/hpccsystems/dfs/client/BinaryRecordReader.java

@@ -75,6 +75,7 @@ public int read() throws IOException
         {
             streamPos++;
         }
+
         return ret;
     }
 
@@ -737,6 +738,15 @@ private void readIntoScratchBuffer(int offset, int dataLen) throws IOException
                 throw e;
             }
 
+            if (bytesRead == 0)
+            {
+                try
+                {
+                    Thread.sleep(1);
+                }
+                catch (InterruptedException e) {}
+            }
+
             position += bytesRead;
             bytesConsumed += bytesRead;
         }
@@ -1294,26 +1304,10 @@ else if ((this.scratchBuffer[strByteLen + bytesScanned] & 0xF8) == 0xF0)
                     // Use the second half of the remaining buffer space as a temp place to read in compressed bytes.
                     // Beginning of the buffer will be used to construct the string
 
-                    int bytesToRead = compressedLen;
-                    int availableBytes = 0;
-                    try
-                    {
-                        availableBytes = this.inputStream.available();
-                    }
-                    catch(Exception e)
-                    {
-                        throw new IOException("Error, unexpected EOS while constructing QString.");
-                    }
-
-                    if (bytesToRead > availableBytes)
-                    {
-                        bytesToRead = availableBytes;
-                    }
-
                     // Scratch buffer is divided into two parts. First expandedLen bytes are for the final expanded string
                     // Remaining bytes are for reading in the compressed string.
                     int readPos = expandedLen + compressedBytesConsumed;
-                    readIntoScratchBuffer(readPos, bytesToRead);
+                    readIntoScratchBuffer(readPos, compressedLen);
 
                     // We want to consume only a whole chunk so round off residual chars
                     // Below we will handle any residual bytes. (strLen % 4)
@@ -1334,7 +1328,7 @@ else if ((this.scratchBuffer[strByteLen + bytesScanned] & 0xF8) == 0xF0)
                         compressedBytesConsumed += QSTR_COMPRESSED_CHUNK_LEN;
                     }
 
-                    compressedBytesRead += bytesToRead;
+                    compressedBytesRead += compressedLen;
                     strByteLen += writePos;
                 }
 

dfsclient/src/main/java/org/hpccsystems/dfs/client/CircularByteBuffer.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * HPCC SYSTEMS software Copyright (C) 2024 HPCC Systems®.
+ * HPCC SYSTEMS software Copyright (C) 2025 HPCC Systems®.
  *
  * Licensed 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
@@ -13,34 +13,58 @@
 
 package org.hpccsystems.dfs.client;
 
-import java.io.IOException;
-
 public class CircularByteBuffer
 {
     private final byte[] buffer;
     private int readPos = 0;
     private int writePos = 0;
-    private int markPos = 0;
+    private int markPos = -1;
     private int bytesReadAfterMark = 0;
     private int currentNumberOfBytes = 0;
 
-    public CircularByteBuffer(int bufferSize)
+    /**
+     * Instantiates a new circular byte buffer.
+     * 
+     * @param bufferSize the buffer size
+     * @throws IllegalArgumentException if buffer size is less than or equal to 0
+     */
+    public CircularByteBuffer(int bufferSize) throws IllegalArgumentException
     {
+        if (bufferSize <= 0)
+        {
+            throw new IllegalArgumentException("Buffer size must be greater than 0");
+        }
+
         buffer = new byte[bufferSize];
     }
 
-    public int getCurrentNumberOfBytes()
+    /**
+     * Gets the number of bytes available in the buffer.
+     * 
+     * @return aviailable bytes
+     */
+    public int getBytesAvailable()
     {
         // We only adjust for the mark internally and when providing information about available space
         return currentNumberOfBytes;
     }
 
+    /**
+     * Checks if the buffer has space.
+     * 
+     * @return true, if successful
+     */
     public boolean hasSpace()
     {
-        return getSpace() > 0;
+        return getFreeSpace() > 0;
     }
 
-    public int getSpace()
+    /**
+     * Gets the free space in the buffer.
+     * 
+     * @return the free space 
+     */
+    public int getFreeSpace()
     {
         int adjustedByteCount = currentNumberOfBytes;
         if (markPos >= 0)
@@ -51,7 +75,12 @@ public int getSpace()
         return buffer.length - adjustedByteCount;
     }
 
-    public int getContiguousSpace()
+    /**
+     * Gets the contiguous free space in the buffer.
+     * 
+     * @return the contiguous free space
+     */
+    public int getContiguousFreeSpace()
     {
         if (!hasSpace())
         {
@@ -75,11 +104,22 @@ public int getContiguousSpace()
         }
     }
 
+    /**
+     * Gets the location of the next write.
+     * 
+     * @return the write offset
+     */
     public int getWriteOffset()
     {
         return writePos;
     }
 
+    /**
+     * Increments write offset.
+     * 
+     * @param increment number of bytes to increment
+     * @return the number of bytes incremented
+     */
     public int incrementWriteOffset(int increment)
     {
         int maxIncrement = buffer.length - writePos;
@@ -95,29 +135,43 @@ public int incrementWriteOffset(int increment)
         return increment;
     }
 
-    public void add(final byte[] targetBuffer, final int offset, final int length) throws IOException 
+    /**
+     * Adds the bytes to the buffer.
+     * 
+     * @param srcBuffer the source buffer
+     * @param offset the offset within the source buffer
+     * @param length the length of bytes to add
+     * @return the number of bytes added
+     */
+    public int add(final byte[] srcBuffer, int offset, int length)
     {
         if (currentNumberOfBytes + length > buffer.length)
         {
-            throw new IOException("Not enough space available");
+            length = buffer.length - currentNumberOfBytes;
         }
 
         for (int i = 0; i < length; i++)
         {
-            buffer[writePos] = targetBuffer[offset + i];
+            buffer[writePos] = srcBuffer[offset + i];
             if (++writePos == buffer.length)
             {
                 writePos = 0;
             }
         }
         currentNumberOfBytes += length;
+        return length;
     }
 
-    public byte read() throws IOException
+    /**
+     * Reads a byte from the buffer.
+     * 
+     * @return the byte read as an int [0-255] or -1 if no bytes are available
+     */
+    public int read()
     {
         if (currentNumberOfBytes <= 0)
         {
-            throw new IOException("No bytes available to read");
+            return -1;
         }
 
         byte b = buffer[readPos];
@@ -128,20 +182,28 @@ public byte read() throws IOException
             bytesReadAfterMark++;
         }
 
-        readPos++;
-        if (readPos >= buffer.length)
+        if (++readPos >= buffer.length)
         {
             readPos = 0;
         }
 
-        return b;
+        int ret = b;
+        return ret + 128;
     }
 
-    public void read(final byte[] targetBuffer, final int targetOffset, final int length) throws IOException
+    /**
+     * Reads bytes from the buffer.
+     * 
+     * @param targetBuffer the target buffer to write to
+     * @param targetOffset the target offset within the target buffer
+     * @param length the number of bytes to read
+     * @return the number of bytes read
+     */
+    public int read(final byte[] targetBuffer, int targetOffset, int length)
     {
         if (length > currentNumberOfBytes)
         {
-            throw new IOException("Not enough bytes available to read");
+            length = currentNumberOfBytes;
         }
 
         if (readPos + length <= buffer.length)
@@ -166,13 +228,26 @@ public void read(final byte[] targetBuffer, final int targetOffset, final int le
         {
             bytesReadAfterMark += length;
         }
+
+        return length;
     }
 
+    /**
+     * Gets the internal buffer.
+     * 
+     * @return the internal buffer
+     */
     public byte[] getInternalBuffer()
     {
         return buffer;
     }
 
+    /**
+     * Marks the current read position, allowing a reset to return to this position.
+     * 
+     * @param readLim the read limit before a reset is no longer allowed
+     * @throws IllegalArgumentException if read limit exceeds available bytes
+     */
     public void mark(int readLim) throws IllegalArgumentException
     {
         if (readLim > buffer.length)
@@ -184,6 +259,9 @@ public void mark(int readLim) throws IllegalArgumentException
         bytesReadAfterMark = 0;
     }
 
+    /**
+     * Resets the read position to the last marked position.
+     */
     public void reset()
     {
         if (markPos < 0)
@@ -198,11 +276,17 @@ public void reset()
         bytesReadAfterMark = 0;
     }
 
-    public long skip(long n) throws IOException
+    /**
+     * Skips the specified number of bytes.
+     * 
+     * @param n the number of bytes to skip
+     * @return the number of bytes skipped
+     */
+    public int skip(int n)
     {
         if (n > currentNumberOfBytes)
         {
-            throw new IOException("Not enough bytes available to skip");
+            n = currentNumberOfBytes;
         }
 
         readPos += n;
@@ -212,6 +296,10 @@ public long skip(long n) throws IOException
         }
 
         currentNumberOfBytes -= n;
+        if (markPos >= 0)
+        {
+            bytesReadAfterMark += n;
+        }
         return n;
     }
 };