Client: Synchronize addReadBuffer to ensure it's not run concurrently with getRead()

getRead() typically is invoked on the client thread, and when used that way this should not be a concern.
However if you are not on the reader thread there is the potential for it to try and read while a write is being added.  This is a use case we want to continue to support, so this fixes it by locking while the read is being added.

Author: jentfoo

README.md

@@ -9,7 +9,7 @@ Include the litesockets library into your project from maven central:
 <dependency>
 	<groupId>org.threadly</groupId>
 	<artifactId>litesockets</artifactId>
-	<version>4.4</version>
+	<version>4.5</version>
 </dependency>
 ```
 

gradle.properties

@@ -1,3 +1,3 @@
 group = org.threadly
-version = 4.4
+version = 4.5
 threadlyVersion = 5.24

src/main/java/org/threadly/litesockets/Client.java

@@ -40,7 +40,6 @@ public abstract class Client implements Closeable {
   protected final SocketExecuterCommonBase se;
   protected final long startTime = Clock.lastKnownForwardProgressingMillis();
   protected final Object readerLock = new Object();
-  protected final Object writerLock = new Object();
   protected final ClientByteStats stats = new ClientByteStats();
   protected final AtomicBoolean closed = new AtomicBoolean(false);
   protected final ConcurrentLinkedQueue<ClientCloseListener> closerListener = new ConcurrentLinkedQueue<>();
@@ -308,14 +307,18 @@ private void runListener(Runnable listener, boolean invokedOnClientThread) {
    * @param bb the {@link ByteBuffer} to add to the clients readBuffer.
    */
   protected void addReadBuffer(final ByteBuffer bb) {
+    if (! bb.hasRemaining()) {
+      return;
+    }
     addReadStats(bb.remaining());
     se.addReadAmount(bb.remaining());
     int start;
-    int end;
-    start = readBuffers.remaining();
-    readBuffers.add(bb);
-    end = readBuffers.remaining();
-    if(end > 0 && start == 0){
+    // synchronize to ensure readBuffers are not modified by non-client thread getRead call
+    synchronized (readerLock) {
+      start = readBuffers.remaining();
+      readBuffers.add(bb);
+    }
+    if(start == 0){
       callReader(true); // we assume all buffers are added from the clients thread
     }
   }

src/main/java/org/threadly/litesockets/TCPClient.java

@@ -40,6 +40,7 @@ public class TCPClient extends Client {
   private final ReuseableMergedByteBuffers writeBuffers = new ReuseableMergedByteBuffers();
   private final Deque<Pair<Long, SettableListenableFuture<Long>>> writeFutures = new ArrayDeque<>(8);
   private final TCPSocketOptions tso = new TCPSocketOptions();
+  protected final Object writerLock = new Object();
   protected final AtomicBoolean startedConnection = new AtomicBoolean(false);
   protected final SettableListenableFuture<Boolean> connectionFuture;
   protected final SocketChannel channel;