add test on Writable FragmentedSubIO + fix it

Author: lecousin

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

@@ -359,6 +359,7 @@ protected ISynchronizationPoint<IOException> canStartWriting() {
 	protected int writeSync(long pos, ByteBuffer buffer) throws IOException {
 		Iterator<RangeLong> it = fragments.iterator();
 		long p = 0;
+		int total = 0;
 		while (it.hasNext()) {
 			RangeLong r = it.next();
 			long s = r.max - r.min + 1;
@@ -376,9 +377,13 @@ protected int writeSync(long pos, ByteBuffer buffer) throws IOException {
 			} else {
 				len = ((IO.Writable.Seekable)io).writeSync(r.min + start, buffer);
 			}
-			return len;
+			total += len;
+			pos += len;
+			p += s;
+			if (!buffer.hasRemaining())
+				return total;
 		}
-		return 0;
+		return total;
 	}
 
 	protected AsyncWork<Integer, IOException> writeAsync(long pos, ByteBuffer buffer, RunnableWithParameter<Pair<Integer, IOException>> ondone) {
@@ -391,26 +396,40 @@ protected AsyncWork<Integer, IOException> writeAsync(long pos, ByteBuffer buffer
 				p += s;
 				continue;
 			}
-			long start = pos - p;
-			int len = buffer.remaining();
-			if (start + len > s) {
-				int prevLimit = buffer.limit();
-				buffer.limit((int)(prevLimit - ((start + len) - s)));
-				return ((IO.Writable.Seekable)io).writeAsync(r.min + start, buffer,
-				new RunnableWithParameter<Pair<Integer,IOException>>() {
-					@Override
-					public void run(Pair<Integer, IOException> param) {
-						buffer.limit(prevLimit);
-						if (ondone != null) ondone.run(param);
-					}
-				});
-			}
-			return operation(((IO.Writable.Seekable)io).writeAsync(r.min + start, buffer, ondone));
+			AsyncWork<Integer,IOException> sp = new AsyncWork<>();
+			writeAsync(it, r, p, 0, pos, buffer, ondone, sp);
+			return operation(sp);
 		}
 		AsyncWork<Integer,IOException> sp = new AsyncWork<>();
 		if (ondone != null) ondone.run(new Pair<>(Integer.valueOf(0), null));
 		sp.unblockSuccess(Integer.valueOf(0));
 		return sp;
 	}
 
+	protected void writeAsync(
+		Iterator<RangeLong> it, RangeLong r, long p, int done, long pos,
+		ByteBuffer buffer, RunnableWithParameter<Pair<Integer, IOException>> ondone, AsyncWork<Integer,IOException> sp
+	) {
+		long start = pos - p;
+		int len = buffer.remaining();
+		long s = r.max - r.min + 1;
+		if (start + len > s) {
+			int prevLimit = buffer.limit();
+			buffer.limit((int)(prevLimit - ((start + len) - s)));
+			IOUtil.listenOnDone(((IO.Writable.Seekable)io).writeAsync(r.min + start, buffer), (nb) -> {
+				buffer.limit(prevLimit);
+				int i = nb.intValue();
+				if (!buffer.hasRemaining() || !it.hasNext()) {
+					IOUtil.success(Integer.valueOf(i), sp, ondone);
+					return;
+				}
+				writeAsync(it, it.next(), p + s, done + i, pos + i, buffer, ondone, sp);
+			}, sp, ondone);
+			return;
+		}
+		IOUtil.listenOnDone(((IO.Writable.Seekable)io).writeAsync(r.min + start, buffer), (nb) -> {
+			IOUtil.success(Integer.valueOf(nb.intValue() + done), sp, ondone);
+		}, sp, ondone);
+	}
+	
 }

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

@@ -46,7 +46,7 @@ public <T extends IO.Readable.Seekable & IO.Writable.Seekable> void testWriteThe
 			buf.position(0);
 			int nb = io.writeSync(buf);
 			if (nb != testBuf.length)
-				throw new Exception("Write only " + nb + " bytes");
+				throw new Exception("Write only " + nb + " bytes for buffer " + i);
 			if (buf.remaining() > 0)
 				throw new Exception("Buffer not fully consumed by write operation");
 		}

net.lecousin.core/src/test/java/net/lecousin/framework/core/tests/io/TestFragmentedSubIOReadWrite.java

@@ -0,0 +1,43 @@
+package net.lecousin.framework.core.tests.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import net.lecousin.framework.concurrent.Task;
+import net.lecousin.framework.core.test.io.TestFragmented;
+import net.lecousin.framework.core.test.io.TestFragmented.FragmentedFile;
+import net.lecousin.framework.core.test.io.TestReadWrite;
+import net.lecousin.framework.io.FileIO;
+import net.lecousin.framework.io.FragmentedSubIO;
+
+@RunWith(Parameterized.class)
+public class TestFragmentedSubIOReadWrite extends TestReadWrite {
+
+	@Parameters
+	public static Collection<Object[]> parameters() throws IOException {
+		return TestFragmented.generateTestCases();
+	}
+	
+	public TestFragmentedSubIOReadWrite(FragmentedFile f) {
+		super(f.testBuf, f.nbBuf);
+		this.f = f;
+	}
+	
+	private FragmentedFile f;
+
+	@SuppressWarnings("unchecked")
+	@Override
+	protected FragmentedSubIO.ReadWrite openReadWrite() throws Exception {
+		File tmpFile = File.createTempFile("test", "fragmentedsubio.rw");
+		tmpFile.deleteOnExit();
+		FileIO.ReadWrite fio = new FileIO.ReadWrite(tmpFile, Task.PRIORITY_NORMAL);
+		fio.setSizeSync(f.file.length());
+		return new FragmentedSubIO.ReadWrite(fio, f.fragments, true, "fragmented IO");
+	}
+	
+}