few fixes on IO

Author: lecousin

net.lecousin.core/src/main/java/net/lecousin/framework/io/IOUtil.java

@@ -253,6 +253,7 @@ public void ready(Integer result) {
 
 			@Override
 			public void error(IOException error) {
+				if (ondone != null) ondone.run(new Pair<>(null, error));
 				sp.unblockError(error);
 			}
 

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

@@ -262,6 +262,15 @@ public void run() {
 				}
 				buffer.flip();
 				synchronized (PreBufferedReadable.this) {
+					if (firstReadTask.isCancelled()) {
+						if (dataReady != null) {
+							SynchronizationPoint<NoException> dr = dataReady;
+							dataReady = null;
+							dr.unblock();
+						}
+						jpNextRead.joined();
+						return;
+					}
 					Throwable e = firstReadTask.getError();
 					if (singleRead && e == null && firstReadTask.getResult().intValue() < size)
 						e = new IOException("Only " + firstReadTask.getResult().intValue()
@@ -420,7 +429,10 @@ public AsyncWork<Integer,IOException> readAsync(ByteBuffer buffer, RunnableWithP
 			@Override
 			public Integer run() throws IOException, CancelException {
 				if (error != null) throw error;
-				if (buffersReady == null) throw new CancelException("IO Closed");
+				if (buffersReady == null) {
+					if (endReached) return Integer.valueOf(-1); // case of empty readable
+					throw new CancelException("IO Closed");
+				}
 				if (current == null) {
 					if (endReached) return Integer.valueOf(-1);
 					if (isClosing() || isClosed()) throw new CancelException("IO Closed");

net.lecousin.core/src/main/java/net/lecousin/framework/io/util/LimitWriteOperations.java

@@ -5,6 +5,7 @@
 
 import net.lecousin.framework.collections.TurnArray;
 import net.lecousin.framework.concurrent.synch.AsyncWork;
+import net.lecousin.framework.concurrent.synch.ISynchronizationPoint;
 import net.lecousin.framework.concurrent.synch.SynchronizationPoint;
 import net.lecousin.framework.exception.NoException;
 import net.lecousin.framework.io.IO;
@@ -28,6 +29,7 @@ public LimitWriteOperations(IO.Writable io, int maxOperations) {
 	private IO.Writable io;
 	private TurnArray<Pair<ByteBuffer,AsyncWork<Integer,IOException>>> waiting;
 	private SynchronizationPoint<NoException> lock = null;
+	private AsyncWork<Integer, IOException> lastWrite = new AsyncWork<>(Integer.valueOf(0), null);
 
 	/**
 	 * Queue the buffer to write. If there is no pending write, the write operation is started.
@@ -37,8 +39,8 @@ public LimitWriteOperations(IO.Writable io, int maxOperations) {
 	public AsyncWork<Integer,IOException> write(ByteBuffer buffer) throws IOException {
 		do {
 			synchronized (waiting) {
-				if (waiting.isEmpty()) {
-					return io.writeAsync(buffer, new RunnableWithParameter<Pair<Integer,IOException>>() {
+				if (waiting.isEmpty() && lastWrite.isUnblocked()) {
+					return lastWrite = io.writeAsync(buffer, new RunnableWithParameter<Pair<Integer,IOException>>() {
 						@Override
 						public void run(Pair<Integer, IOException> param) {
 							writeDone();
@@ -63,12 +65,12 @@ private void writeDone() {
 		synchronized (waiting) {
 			Pair<ByteBuffer,AsyncWork<Integer,IOException>> b = waiting.pollFirst();
 			if (b != null) {
-				io.writeAsync(b.getValue1(), new RunnableWithParameter<Pair<Integer,IOException>>() {
+				(lastWrite = io.writeAsync(b.getValue1(), new RunnableWithParameter<Pair<Integer,IOException>>() {
 					@Override
 					public void run(Pair<Integer, IOException> param) {
 						writeDone();
 					}
-				}).listenInline(b.getValue2());
+				})).listenInline(b.getValue2());
 				if (lock != null) {
 					sp = lock;
 					lock = null;
@@ -81,10 +83,17 @@ public void run(Pair<Integer, IOException> param) {
 
 	/** Return the last pending operation, or null. */
 	public AsyncWork<Integer, IOException> getLastPendingOperation() {
-		Pair<ByteBuffer,AsyncWork<Integer,IOException>> b = waiting.pollLast();
+		Pair<ByteBuffer,AsyncWork<Integer,IOException>> b = waiting.peekLast();
 		if (b == null)
-			return null;
+			return lastWrite.isUnblocked() ? null : lastWrite;
 		return b.getValue2();
 	}
+	
+	/** Same as getLastPendingOperation but never return null (return an unblocked synchronization point instead). */
+	public ISynchronizationPoint<IOException> flush() {
+		ISynchronizationPoint<IOException> sp = getLastPendingOperation();
+		if (sp == null) sp = new SynchronizationPoint<>(true);
+		return sp;
+	}
 
 }

net.lecousin.core/src/test/java/net/lecousin/framework/core/test/io/TestReadableSeekable.java

@@ -215,6 +215,10 @@ public void run() {
 					sp.error(read.get().getError());
 					return;
 				}
+				if (read.get().isCancelled()) {
+					sp.cancel(read.get().getCancelEvent());
+					return;
+				}
 				if (!onDoneBefore.get()) {
 					sp.error(new Exception("Method readFullyAsync didn't call ondone before listeners"));
 					return;

net.lecousin.core/src/test/java/net/lecousin/framework/core/tests/io/util/TestLimitWriteOperations.java

@@ -0,0 +1,39 @@
+package net.lecousin.framework.core.tests.io.util;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+
+import net.lecousin.framework.concurrent.Task;
+import net.lecousin.framework.core.test.LCCoreAbstractTest;
+import net.lecousin.framework.io.FileIO;
+import net.lecousin.framework.io.util.LimitWriteOperations;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestLimitWriteOperations extends LCCoreAbstractTest {
+
+	@Test
+	public void test() throws Exception {
+		File tmp = File.createTempFile("test", "lwo");
+		tmp.deleteOnExit();
+		FileIO.WriteOnly io = new FileIO.WriteOnly(tmp, Task.PRIORITY_NORMAL);
+		LimitWriteOperations writeOps = new LimitWriteOperations(io, 3);
+		byte[] data = new byte[100];
+		for (int i = 0; i < data.length; ++i)
+			data[i] = (byte)i;
+		for (int i = 0; i < 500; ++i)
+			writeOps.write(ByteBuffer.wrap(data));
+		writeOps.flush().blockThrow(0);
+		io.close();
+		FileIO.ReadOnly in = new FileIO.ReadOnly(tmp, Task.PRIORITY_NORMAL);
+		byte[] buf = new byte[data.length];
+		for (int i = 0; i < 500; ++i) {
+			Assert.assertEquals(data.length, in.readFullySync(ByteBuffer.wrap(buf)));
+			Assert.assertArrayEquals(data, buf);
+		}
+		Assert.assertTrue(in.readFullySync(ByteBuffer.wrap(buf)) <= 0);
+		in.close();
+	}
+	
+}