fix BufferedIO + allow to configure LCCore before to start it

Author: lecousin

net.lecousin.core/src/main/java/net/lecousin/framework/application/Application.java

@@ -280,7 +280,7 @@ public static ISynchronizationPoint<Exception> start(
 		app.loggerFactory = new LoggerFactory(app);
 
 		// init LCCore with this application
-		LCCore.init(app);
+		LCCore.start(app);
 
 		JoinPoint<Exception> loading = new JoinPoint<>();
 

net.lecousin.core/src/main/java/net/lecousin/framework/application/LCCore.java

@@ -64,17 +64,29 @@ public static interface Environment {
 	}
 
 	private static Environment instance = null;
+	private static boolean started = false;
 
 	/** Return the environment. */
 	public static Environment get() {
 		return instance;
 	}
 
-	/** Initialization. */
-	public static synchronized void init(Environment env) {
+	/** Set the environment. */
+	public static void set(Environment env) throws IllegalStateException {
 		if (instance != null) throw new IllegalStateException("LCCore environment already exists");
 		instance = env;
-
+	}
+	
+	/** Return true if the environment is already started. */
+	public static boolean isStarted() {
+		return started;
+	}
+	
+	/** Initialization. */
+	public static synchronized void start() {
+		if (started) throw new IllegalStateException("LCCore environment already started");
+		started = true;
+		
 		// init logging system if not specified
 		if (System.getProperty("org.apache.commons.logging.Log") == null) {
 			try {
@@ -102,13 +114,14 @@ public Void run() {
 		}.start();
 	}
 
-	static synchronized void init(Application app) {
+	static synchronized void start(Application app) {
 		if (instance == null) {
 			StandaloneLCCore env = new StandaloneLCCore();
-			env.add(app);
-			init(env);
-		} else
-			instance.add(app);
+			set(env);
+		}
+		instance.add(app);
+		if (!isStarted())
+			start();
 	}
 
 	public static Application getApplication() {

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

@@ -63,7 +63,7 @@ protected BufferedIO(
 		this.bufferSize = bufferSize;
 		this.size = size;
 		this.startReadSecondBufferWhenFirstBufferFullyRead = startReadSecondBufferWhenFirstBufferFullyRead;
-		int nbBuffers = (int)((size - firstBufferSize) / bufferSize) + 2;
+		int nbBuffers = size <= firstBufferSize ? 1 : (int)((size - firstBufferSize) / bufferSize) + 2;
 		if (nbBuffers <= 10)
 			buffers = new ArrayList<>(nbBuffers);
 		else {
@@ -128,6 +128,7 @@ public void run() {
 						return;
 					}
 					buffer.len = loading.getResult().intValue();
+					if (buffer.len < 0) buffer.len = 0;
 					buffer.lastRead = System.currentTimeMillis();
 					manager.newBuffer(buffer);
 					buffer.loading.unblock();
@@ -1032,12 +1033,17 @@ public void run() {
 			while (firstIndex > buffers.size())
 				buffers.add(new Buffer(this));
 			while (firstIndex <= lastIndex) {
-				Buffer b = new Buffer(this);
-				b.buffer = new byte[firstIndex > 0 ? bufferSize : firstBufferSize];
+				Buffer b;
+				if (firstIndex >= buffers.size()) {
+					b = new Buffer(this);
+					newBuffers.add(b);
+					buffers.add(b);
+				} else
+					b = buffers.get(firstIndex);
+				if (b.buffer == null)
+					b.buffer = new byte[firstIndex > 0 ? bufferSize : firstBufferSize];
 				b.len = firstIndex < lastIndex ? b.buffer.length : (int)(newSize - getBufferStart(firstIndex));
 				b.lastRead = System.currentTimeMillis();
-				newBuffers.add(b);
-				buffers.add(b);
 				firstIndex++;
 			}
 		}
@@ -1259,7 +1265,7 @@ protected int writeSync(long pos, ByteBuffer buf, int alreadyDone) throws IOExce
 		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) return alreadyDone + len;
 		return writeSync(p, buf, alreadyDone + len);
@@ -1308,7 +1314,7 @@ protected void write(byte[] buf, int offset, int length) throws IOException {
 			if (pos < size && (bufferIndex != 0 || startReadSecondBufferWhenFirstBufferFullyRead)) {
 				int nextIndex = getBufferIndex(pos);
 				if (nextIndex != bufferIndex) loadBuffer(nextIndex);
-			} else
+			} else if (pos > size)
 				size = pos;
 		} while (length > 0);
 	}
@@ -1351,7 +1357,7 @@ protected void write(byte b) throws IOException {
 		if (pos < size && (bufferIndex != 0 || startReadSecondBufferWhenFirstBufferFullyRead)) {
 			int nextIndex = getBufferIndex(pos);
 			if (nextIndex != bufferIndex) loadBuffer(nextIndex);
-		} else
+		} else if (pos > size)
 			size = pos;
 	}
 }

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

@@ -34,7 +34,7 @@ private BufferingManager() {
 		background.start();
 	}
 
-	static synchronized BufferingManager get() {
+	public static synchronized BufferingManager get() {
 		Application app = LCCore.getApplication();
 		BufferingManager bm = app.getInstance(BufferingManager.class);
 		if (bm == null) {

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

@@ -291,6 +291,9 @@ public void run() {
 						read += buffer.remaining();
 						if (size > 0 && read == size) 
 							endReached = true;
+						if (endReached && size > 0 && read < size) {
+							error = new IOException("Unexpected end after " + read + " bytes read, known size is " + size);
+						}
 						if (dataReady != null) {
 							SynchronizationPoint<NoException> dr = dataReady;
 							dataReady = null;
@@ -499,6 +502,8 @@ public long skipSync(long n) throws IOException {
 					if (reusableBuffers == null) {
 						// we reach the end
 						endReached = true;
+						if (size > 0 && read < size)
+							error = new IOException("Unexpected end after " + read + " bytes read, known size is " + size);
 						if (dataReady != null) dataReady.unblock();
 						return skipped;
 					}
@@ -509,7 +514,9 @@ public long skipSync(long n) throws IOException {
 					if (endReached) return skipped;
 					if (nextReadTask == null && reusableBuffers != null) {
 						// we cancelled all buffers, let's do a move
-						skipped += src.skipSync(n);
+						long n2 = src.skipSync(n);
+						skipped += n2;
+						read += n2;
 						// restart reading
 						stopReading = false;
 						moveNextBuffer(true);
@@ -629,6 +636,8 @@ public void run() {
 							if (!endReached && !reusableBuffers.isEmpty() && !stopReading)
 								nextRead();
 						}
+						if (endReached && size > 0 && read < size)
+							error = new IOException("Unexpected end after " + read + " bytes read, known size is " + size);
 						if (dataReady != null) {
 							sp = dataReady;
 							dataReady = null;

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

@@ -54,7 +54,7 @@ public ReadableToSeekable(IO.Readable io, int bufferSize) throws IOException {
 		file.deleteOnExit();
 		@SuppressWarnings("resource")
 		FileIO.ReadWrite fio = new FileIO.ReadWrite(file, io.getPriority());
-		buffered = new BufferedIO.ReadWrite(fio, 512, bufferSize, 0L, true);
+		buffered = new BufferedIO.ReadWrite(fio, 512, bufferSize, 0L, false);
 		readNextBuffer();
 	}
 

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

@@ -4,10 +4,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
-import org.junit.Assert;
-import org.junit.Assume;
-import org.junit.Test;
-
 import net.lecousin.framework.collections.ArrayUtil;
 import net.lecousin.framework.concurrent.synch.AsyncWork;
 import net.lecousin.framework.concurrent.synch.SynchronizationPoint;
@@ -21,6 +17,10 @@
 import net.lecousin.framework.util.Pair;
 import net.lecousin.framework.util.RunnableWithParameter;
 
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
 public abstract class TestReadable extends TestIO.UsingGeneratedTestFiles {
 
 	protected TestReadable(File testFile, byte[] testBuf, int nbBuf) {

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

@@ -122,7 +122,7 @@ public void run() {
 					return;
 				}
 				if (read.get().getResult().intValue() != 1) {
-					sp.error(new Exception("Unexpected end of stream at " + (offset.get()*testBuf.length+j.get())));
+					sp.error(new Exception("Unexpected end of stream at " + (offset.get()*testBuf.length+j.get()) + ": 1 byte expected, " + read.get().getResult().intValue() + " read"));
 					return;
 				}
 				if (b[0] != testBuf[j.get()]) {