Merge branch 'wip/codecs' of github.com:saalfeldlab/n5 into wip/codecs

Author: bogovicj

doc/README.md

@@ -132,6 +132,6 @@ Custom compression schemes can be implemented using the annotation discovery mec
 
 ## Disclaimer
 
-HDF5 is a great format that provides a wealth of conveniences that I do not want to miss.  It's inefficiency for parallel writing, however, limit its applicability for handling of very large n-dimensional data.
+HDF5 is a great format that provides a wealth of conveniences that I do not want to miss.  Its inefficiency for parallel writing, however, limits its applicability for handling of very large n-dimensional data.
 
 N5 uses the native filesystem of the target platform and JSON files to specify basic and custom meta-data as attributes.  It aims at preserving the convenience of HDF5 where possible but doesn't try too hard to be a full replacement.

pom.xml

@@ -202,10 +202,6 @@
 			<groupId>org.apache.commons</groupId>
 			<artifactId>commons-compress</artifactId>
 		</dependency>
-		<dependency>
-			<groupId>commons-io</groupId>
-			<artifactId>commons-io</artifactId>
-		</dependency>
 
 		<!-- JMH -->
 		<dependency>

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

@@ -65,6 +65,7 @@
 import java.io.Writer;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.OverlappingFileLockException;
@@ -85,6 +86,8 @@
 import java.util.Iterator;
 import java.util.stream.Stream;
 
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
+
 /**
  * Filesystem {@link KeyValueAccess}.
  *
@@ -145,6 +148,10 @@ protected LockedFileChannel(final Path path, final boolean readOnly) throws IOEx
 			}
 		}
 
+		protected FileChannel getFileChannel() {
+			return channel;
+		}
+
 		private void truncateChannel(int size) {
 
 			try {
@@ -202,7 +209,12 @@ public FileSystemKeyValueAccess(final FileSystem fileSystem) {
 	}
 
 	@Override
-	public LockedChannel lockForReading(final String normalPath) throws N5IOException {
+	public ReadData createReadData(final String normalPath) {
+		return new KeyValueAccessReadData(new FileLazyRead(normalPath));
+	}
+
+	@Override
+	public LockedFileChannel lockForReading(final String normalPath) throws N5IOException {
 
 		try {
 			return new LockedFileChannel(normalPath, true);
@@ -268,6 +280,18 @@ public boolean exists(final String normalPath) {
 		return Files.exists(path);
 	}
 
+	@Override
+	public long size(final String normalPath) {
+
+		try {
+			return Files.size(fileSystem.getPath(normalPath));
+		} catch (NoSuchFileException e) {
+			throw new N5Exception.N5NoSuchKeyException("No such file", e);
+		} catch (IOException | UncheckedIOException e) {
+			throw new N5Exception.N5IOException(e);
+		}
+	}
+
 	@Override
 	public String[] listDirectories(final String normalPath) throws N5IOException {
 
@@ -595,4 +619,58 @@ protected static void createAndCheckIsDirectory(
 				throw x;
 		}
 	}
+
+	private class FileLazyRead implements LazyRead {
+
+		private final String normalKey;
+
+	    FileLazyRead(String normalKey) {
+	        this.normalKey = normalKey;
+	    }
+
+	    @Override
+	    public long size() {
+	        return FileSystemKeyValueAccess.this.size(normalKey);
+	    }
+
+	    @Override
+	    public ReadData materialize(final long offset, final long length) {
+
+	        try (final LockedFileChannel lfs = new LockedFileChannel(normalKey, true)) {
+	            final FileChannel channel = lfs.getFileChannel();
+	            channel.position(offset);
+	            if (length > Integer.MAX_VALUE)
+	                throw new IOException("Attempt to materialize too large data");
+
+	            final long channelSize = channel.size();
+	            if (!validBounds(channelSize, offset, length))
+	                throw new IndexOutOfBoundsException();
+
+	            final int sz = (int) (length < 0 ? channelSize : length);
+	            final byte[] data = new byte[sz];
+	            final ByteBuffer buf = ByteBuffer.wrap(data);
+	            channel.read(buf);
+	            return ReadData.from(data);
+
+	        } catch (final NoSuchFileException e) {
+	            throw new N5NoSuchKeyException("No such file", e);
+	        } catch (IOException | UncheckedIOException e) {
+	            throw new N5Exception.N5IOException(e);
+	        }
+	    }
+
+	}
+
+	private static boolean validBounds(long channelSize, long offset, long length) {
+
+		if (offset < 0)
+			return false;
+		else if (channelSize > 0 && offset >= channelSize) // offset == 0 and arrayLength == 0 is okay
+			return false;
+		else if (length >= 0 && offset + length > channelSize)
+			return false;
+
+		return true;
+	}
+
 }

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

@@ -29,8 +29,9 @@
 package org.janelia.saalfeldlab.n5;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.io.UncheckedIOException;
-import java.util.Arrays;
 
 import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
 import org.janelia.saalfeldlab.n5.codec.Codec.ArrayCodec;
@@ -81,14 +82,13 @@ default JsonElement getAttributes(final String pathName) throws N5Exception {
 		final String groupPath = N5URI.normalizeGroupPath(pathName);
 		final String attributesPath = absoluteAttributesPath(groupPath);
 
-		try (final LockedChannel lockedChannel = getKeyValueAccess().lockForReading(attributesPath)) {
-			return GsonUtils.readAttributes(lockedChannel.newReader(), getGson());
+		try ( final InputStream in = getKeyValueAccess().createReadData(attributesPath).inputStream() ) {
+			return GsonUtils.readAttributes(new InputStreamReader(in), getGson());
 		} catch (final N5Exception.N5NoSuchKeyException e) {
 			return null;
 		} catch (final IOException | UncheckedIOException | N5IOException e) {
 			throw new N5IOException("Failed to read attributes from dataset " + pathName, e);
 		}
-
 	}
 
 	@Override
@@ -99,15 +99,12 @@ default DataBlock<?> readBlock(
 
 		final String path = absoluteDataBlockPath(N5URI.normalizeGroupPath(pathName), gridPosition);
 
-		try (final LockedChannel lockedChannel = getKeyValueAccess().lockForReading(path)) {
-			final ArrayCodec codec = datasetAttributes.getArrayCodec();
-			return codec.decode(ReadData.from(lockedChannel.newInputStream()), gridPosition);
-		} catch (final N5Exception.N5NoSuchKeyException e) {
+		try {
+			final ReadData blockData = getKeyValueAccess().createReadData(path);
+			final ArrayCodec arrayCodec = datasetAttributes.getArrayCodec();
+			return arrayCodec.decode(blockData, gridPosition);
+		} catch (N5Exception.N5NoSuchKeyException e) {
 			return null;
-		} catch (final IOException | UncheckedIOException | N5IOException e) {
-			throw new N5IOException(
-					"Failed to read block " + Arrays.toString(gridPosition) + " from dataset " + path,
-					e);
 		}
 	}
 

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

@@ -33,8 +33,10 @@
 import org.apache.commons.lang3.function.TriFunction;
 import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
 import org.janelia.saalfeldlab.n5.http.ListResponseParser;
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
 
 import java.io.Closeable;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
@@ -135,6 +137,12 @@ public boolean exists(final String normalPath) {
 		}
 	}
 
+	@Override public long size(String normalPath) {
+
+		final HttpURLConnection head = requireValidHttpResponse(normalPath, "HEAD", "Error checking existence: " + normalPath, true);
+		return head.getContentLengthLong();
+	}
+
 	/**
 	 * Test whether the path is a directory.
 	 * <p>
@@ -224,12 +232,16 @@ private HttpURLConnection httpRequest(String normalPath, String method) throws I
 	}
 
 	@Override
+	public ReadData createReadData(final String normalPath) {
+		return new KeyValueAccessReadData(new HttpLazyRead(normalPath));
+	}
+
 	public LockedChannel lockForReading(final String normalPath) throws N5IOException {
 		//TODO Caleb: Maybe check exists lazily when attempting to read
 		try {
 			if (!exists(normalPath))
 				throw new N5Exception.N5NoSuchKeyException("Key does not exist: " + normalPath);
-			return new HttpObjectChannel(uri(normalPath));
+			return new HttpObjectChannel(uri(normalPath), 0, -1);
 		} catch (URISyntaxException e) {
 			throw new N5Exception("Invalid URI Syntax", e);
 		}
@@ -341,23 +353,49 @@ public void delete(final String normalPath) {
 	private class HttpObjectChannel implements LockedChannel {
 
 		protected final URI uri;
+		private final long startByte;
+		private final long size;
 		private final ArrayList<Closeable> resources = new ArrayList<>();
 
-		protected HttpObjectChannel(final URI uri) {
+		protected HttpObjectChannel(final URI uri, long startByte, long size) {
 
 			this.uri = uri;
+			this.startByte = startByte;
+			this.size = size;
+		}
+
+		private boolean isPartialRead() {
+			return startByte > 0 || (size >= 0 && size != Long.MAX_VALUE);
 		}
 
 		@Override
 		public InputStream newInputStream() throws N5IOException {
 
 			try {
-				return uri.toURL().openStream();
+				HttpURLConnection conn = (HttpURLConnection)uri.toURL().openConnection();
+				if (isPartialRead()) {
+					conn.setRequestProperty(RANGE, rangeString());
+					final String acceptRanges = conn.getHeaderField(ACCEPT_RANGE);
+					if (acceptRanges == null || !acceptRanges.equals(BYTES)) {
+						conn.disconnect();
+						conn = (HttpURLConnection)uri.toURL().openConnection();
+						return ReadData.from(conn.getInputStream()).materialize().slice(startByte, size).inputStream();
+					}
+				}
+				return conn.getInputStream();
+			} catch (FileNotFoundException e) {
+				throw new N5Exception.N5NoSuchKeyException("Could not open stream for " + uri, e);
 			} catch (IOException e) {
 				throw new N5IOException("Could not open stream for " + uri, e);
 			}
 		}
 
+		private String rangeString() {
+
+			final String lastByte = (size > 0) ? Long.toString(startByte + size - 1) : "";
+			return String.format("%s=%d-%s", BYTES, startByte, lastByte);
+		}
+
 		@Override
 		public Reader newReader() throws N5IOException {
 
@@ -392,4 +430,29 @@ public void close() throws IOException {
 		}
 	}
 
+	private class HttpLazyRead implements LazyRead {
+
+		private final String normalKey;
+
+		HttpLazyRead(String normalKey) {
+			this.normalKey = normalKey;
+		}
+
+		@Override
+		public long size() {
+			return HttpKeyValueAccess.this.size(normalKey);
+		}
+
+		@Override
+		public ReadData materialize(long offset, long length) {
+			try (final HttpObjectChannel ch = new HttpObjectChannel(uri(normalKey), offset, length)) {
+				return ReadData.from(ch.newInputStream()).materialize();
+			} catch (IOException e) {
+				throw new N5IOException(e);
+			} catch (URISyntaxException e) {
+				throw new N5Exception(e);
+			}
+		}
+	}
+
 }