add logs when IO not correctly closed

Author: lecousin

net.lecousin.core/src/main/java/net/lecousin/framework/concurrent/synch/AsyncWork.java

@@ -1,6 +1,7 @@
 package net.lecousin.framework.concurrent.synch;
 
 import java.util.ArrayList;
+import java.util.Collection;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import net.lecousin.framework.application.Application;
@@ -72,6 +73,12 @@ public boolean hasError() {
 		return error != null;
 	}
 
+	@Override
+	public Collection<?> getAllListeners() {
+		if (listenersInline == null) return new ArrayList<>(0);
+		return new ArrayList<>(listenersInline);
+	}
+	
 	/** Add a listener to be called when this AsyncWork is unblocked. */
 	public void listenInline(AsyncWorkListener<T,TError> listener) {
 		synchronized (this) {

net.lecousin.core/src/main/java/net/lecousin/framework/concurrent/synch/ISynchronizationPoint.java

@@ -1,5 +1,7 @@
 package net.lecousin.framework.concurrent.synch;
 
+import java.util.Collection;
+
 import net.lecousin.framework.concurrent.CancelException;
 import net.lecousin.framework.concurrent.Task;
 import net.lecousin.framework.event.Listener;
@@ -230,4 +232,7 @@ default AsyncWork<Void, TError> toAsyncWorkVoid() {
 		return aw;
 	}
 
+	/** Return a collection with all listeners, only for debugging purposes. */
+	public Collection<?> getAllListeners();
+	
 }

net.lecousin.core/src/main/java/net/lecousin/framework/concurrent/synch/LockPoint.java

@@ -1,6 +1,7 @@
 package net.lecousin.framework.concurrent.synch;
 
 import java.util.ArrayList;
+import java.util.Collection;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import net.lecousin.framework.concurrent.BlockedThreadHandler;
@@ -92,6 +93,12 @@ public void blockPause(long logAfter) {
 		}
 	}
 
+	@Override
+	public Collection<?> getAllListeners() {
+		if (listeners == null) return new ArrayList<>(0);
+		return new ArrayList<>(listeners);
+	}
+
 	@Override
 	public void listenInline(Runnable r) {
 		synchronized (this) {

net.lecousin.core/src/main/java/net/lecousin/framework/concurrent/synch/SynchronizationPoint.java

@@ -1,6 +1,7 @@
 package net.lecousin.framework.concurrent.synch;
 
 import java.util.ArrayList;
+import java.util.Collection;
 
 import net.lecousin.framework.application.Application;
 import net.lecousin.framework.application.LCCore;
@@ -41,6 +42,12 @@ public SynchronizationPoint(CancelException cancel) {
 	private CancelException cancelled = null;
 	private ArrayList<Runnable> listenersInline = null;
 
+	@Override
+	public Collection<?> getAllListeners() {
+		if (listenersInline == null) return new ArrayList<>(0);
+		return new ArrayList<>(listenersInline);
+	}
+	
 	@Override
 	public void listenInline(Runnable r) {
 		synchronized (this) {

net.lecousin.core/src/main/java/net/lecousin/framework/concurrent/synch/WaitingDataQueueSynchronizationPoint.java

@@ -1,6 +1,7 @@
 package net.lecousin.framework.concurrent.synch;
 
 import java.util.ArrayList;
+import java.util.Collection;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import net.lecousin.framework.collections.TurnArray;
@@ -93,6 +94,12 @@ public void endOfData() {
 		}
 	}
 
+	@Override
+	public Collection<?> getAllListeners() {
+		if (listeners == null) return new ArrayList<>(0);
+		return new ArrayList<>(listeners);
+	}
+	
 	@Override
 	public boolean isUnblocked() {
 		return !waitingData.isEmpty() || cancel != null || error != null;

net.lecousin.core/src/main/java/net/lecousin/framework/io/buffering/BufferedIO.java

@@ -1189,7 +1189,7 @@ public Void run() {
 				if (p < size && (bufferIndex != 0 || startReadSecondBufferWhenFirstBufferFullyRead)) {
 					int nextIndex = getBufferIndex(p);
 					if (nextIndex != bufferIndex) loadBuffer(nextIndex);
-				} else
+				} else if (p > size)
 					size = p;
 				if (buf.remaining() == 0) {
 					if (ondone != null) ondone.run(new Pair<>(Integer.valueOf(len + alreadyDone), null));

net.lecousin.core/src/main/java/net/lecousin/framework/util/ConcurrentCloseable.java

@@ -149,8 +149,11 @@ public ISynchronizationPoint<Exception> closeAsync() {
 			StringBuilder s = new StringBuilder();
 			s.append("Closeable still waiting for pending operations: ").append(this);
 			for (ISynchronizationPoint<?> op : pending)
-				if (!op.isUnblocked())
+				if (!op.isUnblocked()) {
 					s.append("\r\n - ").append(op);
+					for (Object o : op.getAllListeners())
+						s.append("\r\n    - ").append(o);
+				}
 			if (underlying != null && !underlying.isUnblocked())
 				s.append("\r\n - closeUnderlyingResources");
 			LCCore.getApplication().getDefaultLogger().error(s.toString());