perf: Improved decoding from InputStreams

src/main/java/org/janelia/saalfeldlab/n5/codec/FlatArrayCodec.java

@@ -30,6 +30,7 @@
 
 import java.io.DataInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.IntBuffer;
@@ -131,8 +132,8 @@ public ReadData encode(final byte[] data) {
 		@Override
 		public byte[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final byte[] data = newArray(numElements);
-			try {
-				new DataInputStream(readData.inputStream()).readFully(data);
+			try (final InputStream is = readData.limit(numElements).inputStream()) {
+				new DataInputStream(is).readFully(data);
 			} catch (IOException e) {
 				throw new N5IOException(e);
 			}
@@ -159,7 +160,7 @@ public ReadData encode(final short[] data) throws N5IOException {
 		@Override
 		public short[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final short[] data = newArray(numElements);
-			readData.toByteBuffer().order(order).asShortBuffer().get(data);
+			readData.limit(2 * numElements).toByteBuffer().order(order).asShortBuffer().get(data);
 			return data;
 		}
 	}
@@ -183,7 +184,7 @@ public ReadData encode(final int[] data) throws N5IOException {
 		@Override
 		public int[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final int[] data = newArray(numElements);
-			final ByteBuffer byteBuffer = readData.toByteBuffer();
+			final ByteBuffer byteBuffer = readData.limit(4 * numElements).toByteBuffer();
 			final IntBuffer intBuffer = byteBuffer.order(order).asIntBuffer();
 			intBuffer.get(data);
 			return data;
@@ -209,7 +210,7 @@ public ReadData encode(final long[] data) throws N5IOException {
 		@Override
 		public long[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final long[] data = newArray(numElements);
-			readData.toByteBuffer().order(order).asLongBuffer().get(data);
+			readData.limit(8 * numElements).toByteBuffer().order(order).asLongBuffer().get(data);
 			return data;
 		}
 	}
@@ -233,7 +234,7 @@ public ReadData encode(final float[] data) throws N5IOException {
 		@Override
 		public float[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final float[] data = newArray(numElements);
-			readData.toByteBuffer().order(order).asFloatBuffer().get(data);
+			readData.limit(4 * numElements).toByteBuffer().order(order).asFloatBuffer().get(data);
 			return data;
 		}
 	}
@@ -257,7 +258,7 @@ public ReadData encode(final double[] data) throws N5IOException {
 		@Override
 		public double[] decode(final ReadData readData, int numElements) throws N5IOException {
 			final double[] data = newArray(numElements);
-			readData.toByteBuffer().order(order).asDoubleBuffer().get(data);
+			readData.limit(8 * numElements).toByteBuffer().order(order).asDoubleBuffer().get(data);
 			return data;
 		}
 	}
@@ -347,8 +348,8 @@ public ReadData encode(byte[] data) throws N5IOException {
 		@Override
 		public byte[] decode(ReadData readData, int numElements) throws N5IOException {
 			final byte[] data = newArray(numElements);
-			try {
-				new DataInputStream(readData.inputStream()).readFully(data);
+			try (final InputStream is = readData.inputStream()) {
+				new DataInputStream(is).readFully(data);
 			} catch (IOException e) {
 				throw new N5IOException(e);
 			}

src/main/java/org/janelia/saalfeldlab/n5/codec/N5BlockCodecs.java

@@ -150,18 +150,24 @@ public ReadData encode(DataBlock<T> dataBlock) throws N5IOException {
 		@Override
 		public DataBlock<T> decode(final ReadData readData, final long[] gridPosition) throws N5IOException {
 
+			// read block header with input stream since header is variable length
+			final BlockHeader header;
 			try(final InputStream in = readData.inputStream()) {
-				final BlockHeader header = decodeBlockHeader(in);
-
-				final int numElements = header.numElements();
-				final ReadData decodeData = codec.decode(ReadData.from(in));
-
-				// the dataCodec knows the number of bytes per element
-				final T data = dataCodec.decode(decodeData, numElements);
-				return dataBlockFactory.createDataBlock(header.blockSize(), gridPosition, data);
+				header = decodeBlockHeader(in);
 			} catch (IOException e) {
 				throw new N5IOException(e);
 			}
+
+			// determine length
+			// and slice original read data so that bodyReadData is known length
+			final int numElements = header.numElements();
+			final long bodyLength = readData.length() - header.getSize();
+			final ReadData bodyReadData = bodyLength > 0 ? readData.slice(header.getSize(), bodyLength) : ReadData.empty();
+			final ReadData decodeData = codec.decode(bodyReadData);
+
+			// the dataCodec knows the number of bytes per element
+			final T data = dataCodec.decode(decodeData, numElements);
+			return dataBlockFactory.createDataBlock(header.blockSize(), gridPosition, data);
 		}
 	}
 
@@ -282,16 +288,7 @@ static class BlockHeader {
 
 		public int getSize() {
 
-			switch (mode) {
-				case MODE_DEFAULT:
-					return 2 + 4 * blockSize.length;
-				case MODE_VARLENGTH:
-					return 2 + 4 * blockSize.length + 4;
-				case MODE_OBJECT:
-					return 2 + 4;
-				default:
-					throw new IllegalArgumentException("Unexpected mode: " + mode);
-			}
+			return headerSizeInBytes(mode, blockSize == null ? 0 : blockSize.length);
 		}
 
 		public int[] blockSize() {

src/main/java/org/janelia/saalfeldlab/n5/readdata/InputStreamReadData.java

@@ -66,6 +66,11 @@ public ReadData slice(final long offset, final long length) throws N5IOException
 		return bytes.slice(offset, length);
 	}
 
+	@Override
+	public ReadData limit(final long length) {
+		return new InputStreamReadData(inputStream, (int)length);
+	}
+
 	private boolean inputStreamCalled = false;
 
 	@Override