Merge pull request #173 from saalfeldlab/segmented-readdata

Add SegmentedReadData (and other ReadData improvements)

Author: bogovicj

src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccessReadData.java

@@ -17,7 +17,7 @@ public class KeyValueAccessReadData implements ReadData {
     private final long offset;
     private long length;
 
-    public KeyValueAccessReadData(LazyRead lazyRead) {
+	public KeyValueAccessReadData(final LazyRead lazyRead) {
         this(lazyRead, 0, -1);
     }
 
@@ -28,11 +28,13 @@ public KeyValueAccessReadData(LazyRead lazyRead) {
     }
 
     @Override
-    public ReadData materialize() throws N5IOException {
-        if (materialized == null)
-            materialized = lazyRead.materialize(offset, length);
-        return materialized;
-    }
+	public ReadData materialize() throws N5IOException {
+		if (materialized == null) {
+			materialized = lazyRead.materialize(offset, length);
+			length = materialized.length();
+		}
+		return this;
+	}
 
 	/**
 	 * Returns a {@link ReadData} whose length is limited to the given value.
@@ -50,39 +52,44 @@ public ReadData materialize() throws N5IOException {
     public ReadData slice(final long offset, final long length) throws N5IOException {
         if (offset < 0)
             throw new IndexOutOfBoundsException("Negative offset: " + offset);
-            
+
         if (materialized != null)
             return materialized.slice(offset, length);
 
         // if a slice of indeterminate length is requested, but the
         // length is already known, use the known length;
-        final int lengthArg;
+        final long lengthArg;
         if (this.length > 0 && length < 0)
-            lengthArg = (int)(this.length - offset);
+            lengthArg = this.length - offset;
         else
-            lengthArg = (int)length;
+            lengthArg = length;
 
         return new KeyValueAccessReadData(lazyRead, this.offset + offset, lengthArg);
     }
 
     @Override
     public InputStream inputStream() throws N5IOException, IllegalStateException {
-        return materialize().inputStream();
+        materialize();
+		return materialized.inputStream();
     }
 
     @Override
     public byte[] allBytes() throws N5IOException, IllegalStateException {
-        return materialize().allBytes();
+		materialize();
+		return materialized.allBytes();
     }
 
     @Override
-    public long length() throws N5IOException {
-        if (materialized != null)
-            return materialized.length();
-        if (length < 0) {
-            length = lazyRead.size() - offset;
-        }
+    public long length() {
         return length;
     }
 
+	@Override
+	public long requireLength() throws N5IOException {
+		if (length < 0) {
+			length = lazyRead.size() - offset;
+		}
+		return length;
+	}
+
 }

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

@@ -65,15 +65,20 @@ public long length() {
 		return length;
 	}
 
+	@Override
+	public long requireLength() {
+		return length;
+	}
+
 	@Override
 	public InputStream inputStream() {
+
 		return new ByteArrayInputStream(data, offset, length);
 	}
 
 	@Override
 	public byte[] allBytes() {
 
-		// alternatively, we could always return the requested length
 		if (offset == 0 && data.length == length) {
 			return data;
 		} else {

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

@@ -6,13 +6,13 @@
  * %%
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
- * 
+ *
  * 1. Redistributions of source code must retain the above copyright notice,
  *    this list of conditions and the following disclaimer.
  * 2. Redistributions in binary form must reproduce the above copyright notice,
  *    this list of conditions and the following disclaimer in the documentation
  *    and/or other materials provided with the distribution.
- * 
+ *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
@@ -28,10 +28,14 @@
  */
 package org.janelia.saalfeldlab.n5.readdata;
 
+import java.io.DataInputStream;
+import java.io.IOException;
 import java.io.InputStream;
+import org.apache.commons.io.IOUtils;
+import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
 
 // not thread-safe
-class InputStreamReadData extends AbstractInputStreamReadData {
+class InputStreamReadData implements ReadData {
 
 	private final InputStream inputStream;
 	private final int length;
@@ -43,18 +47,67 @@ class InputStreamReadData extends AbstractInputStreamReadData {
 
 	@Override
 	public long length() {
-		return length;
+		return (bytes != null) ? bytes.length() : length;
+	}
+
+	@Override
+	public long requireLength() throws N5IOException {
+		long l = length();
+		if ( l >= 0 ) {
+			return l;
+		} else {
+			return materialize().length();
+		}
+	}
+
+	@Override
+	public ReadData slice(final long offset, final long length) throws N5IOException {
+		materialize();
+		return bytes.slice(offset, length);
 	}
 
 	private boolean inputStreamCalled = false;
 
 	@Override
 	public InputStream inputStream() throws IllegalStateException {
-		if (inputStreamCalled) {
-			throw new IllegalStateException("InputStream() already called");
-		} else {
+		if (bytes != null) {
+			return bytes.inputStream();
+		} else if (!inputStreamCalled) {
 			inputStreamCalled = true;
 			return inputStream;
+		} else {
+			throw new IllegalStateException("InputStream() already called");
+		}
+	}
+
+	@Override
+	public byte[] allBytes() throws N5IOException, IllegalStateException {
+		materialize();
+		return bytes.allBytes();
+	}
+
+	private ByteArrayReadData bytes;
+
+	@Override
+	public ReadData materialize() throws N5IOException {
+		if (bytes == null) {
+			final byte[] data;
+			if (length >= 0) {
+				data = new byte[length];
+				try (InputStream is = inputStream()) {
+					new DataInputStream(is).readFully(data);
+				} catch (IOException e) {
+					throw new N5IOException(e);
+				}
+			} else {
+				try (InputStream is = inputStream()) {
+					data = IOUtils.toByteArray(is);
+				} catch (IOException e) {
+					throw new N5IOException(e);
+				}
+			}
+			bytes = new ByteArrayReadData(data);
 		}
+		return this;
 	}
 }

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

@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import org.apache.commons.io.output.CountingOutputStream;
 import org.apache.commons.io.output.ProxyOutputStream;
 import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
 
@@ -42,13 +43,13 @@ class LazyReadData implements ReadData {
 	}
 
 	/**
-	 * Construct a {@code LazyReadData} that uses the given {@code OutputStreamEncoder} to
+	 * Construct a {@code LazyReadData} that uses the given {@code OutputStreamOperator} to
 	 * encode the given {@code ReadData}.
 	 *
 	 * @param data
 	 * 		the ReadData to encode
 	 * @param encoder
-	 * 		OutputStreamEncoder to use for encoding
+	 * 		OutputStreamOperator to use for encoding
 	 */
 	LazyReadData(final ReadData data, final OutputStreamOperator encoder) {
 		this(outputStream -> {
@@ -62,30 +63,53 @@ class LazyReadData implements ReadData {
 
 	private ByteArrayReadData bytes;
 
+	private long length = -1;
+
 	@Override
 	public ReadData materialize() throws N5IOException {
 		if (bytes == null) {
-			final ByteArrayOutputStream baos = new ByteArrayOutputStream(8192);
-			writeTo(baos);
-			bytes = new ByteArrayReadData(baos.toByteArray());
+			try {
+				final ByteArrayOutputStream baos = new ByteArrayOutputStream(8192);
+				writer.writeTo(baos);
+				bytes = new ByteArrayReadData(baos.toByteArray());
+			} catch (IOException e) {
+				throw new N5IOException(e);
+			}
 		}
-		return bytes;
+		return this;
+	}
+
+	@Override
+	public long length() {
+		return (bytes != null) ? bytes.length() : length;
 	}
 
 	@Override
-	public long length() throws N5IOException {
+	public long requireLength() throws N5IOException {
+		long l = length();
+		if ( l >= 0 ) {
+			return l;
+		} else {
+			return materialize().length();
+		}
+	}
 
-		return materialize().length();
+	@Override
+	public ReadData slice(final long offset, final long length) throws N5IOException {
+		materialize();
+		return bytes.slice(offset, length);
 	}
 
 	@Override
 	public InputStream inputStream() throws N5IOException, IllegalStateException {
-		return materialize().inputStream();
+		materialize();
+		return bytes.inputStream();
 	}
 
 	@Override
 	public byte[] allBytes() throws N5IOException, IllegalStateException {
-		return materialize().allBytes();
+		materialize();
+		return bytes.allBytes();
 	}
 
 	@Override
@@ -94,7 +118,9 @@ public void writeTo(final OutputStream outputStream) throws N5IOException, Illeg
 			if (bytes != null) {
 				outputStream.write(bytes.allBytes());
 			} else {
-				writer.writeTo(outputStream);
+				final CountingOutputStream cos = new CountingOutputStream(outputStream);
+				writer.writeTo(cos);
+				length = cos.getByteCount();
 			}
 		} catch (IOException e) {
 			throw new N5IOException(e);
@@ -105,7 +131,7 @@ public void writeTo(final OutputStream outputStream) throws N5IOException, Illeg
 	 * {@code UnaryOperator} that wraps {@code OutputStream} to intercept {@code
 	 * close()} and call {@code flush()} instead
 	 */
-	private static OutputStreamOperator interceptClose = o -> new ProxyOutputStream(o) {
+	private static final OutputStreamOperator interceptClose = o -> new ProxyOutputStream(o) {
 
 		@Override
 		public void close() throws IOException {