Merge pull request #156 from bogovicj/splittable-readdata

Add slice and limit methods to ReadData

Author: cmhulbert

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/HttpKeyValueAccess.java

@@ -29,9 +29,11 @@
 package org.janelia.saalfeldlab.n5;
 
 import org.apache.commons.io.IOUtils;
+
 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.IOException;
@@ -57,6 +59,13 @@
  */
 public class HttpKeyValueAccess implements KeyValueAccess {
 
+	public static final String HEAD = "HEAD";
+	public static final String GET = "GET";
+
+	public static final String RANGE = "Range";
+	public static final String ACCEPT_RANGE = "Accept-Range";
+	public static final String BYTES = "bytes";
+
 	private int readTimeoutMilliseconds;
 	private int connectionTimeoutMilliseconds;
 
@@ -127,6 +136,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>
@@ -142,7 +157,7 @@ public boolean exists(final String normalPath) {
 	public boolean isDirectory(final String normalPath) {
 
 		try {
-			requireValidHttpResponse(getDirectoryPath(normalPath), "HEAD", (code,  msg,http) -> {
+			requireValidHttpResponse(getDirectoryPath(normalPath), HEAD, (code,  msg,http) -> {
 				final N5Exception cause = validExistsResponse(code, "Error checking directory: " + normalPath, msg, true);
 				if (code >= 300 && code < 400) {
 					final String redirectLocation = http.getHeaderField("Location");
@@ -184,7 +199,7 @@ public boolean isFile(final String normalPath) {
 
 		/* Files must not end in `/` And Don't accept a redirect to a location ending in `/` */
 		try {
-			requireValidHttpResponse(getFilePath(normalPath), "HEAD", (code, msg, http) -> {
+			requireValidHttpResponse(getFilePath(normalPath), HEAD, (code, msg, http) -> {
 				final N5Exception cause = validExistsResponse(code, "Error accessing file: " + normalPath, msg, true);
 				if (code >= 300 && code < 400) {
 					final String redirectLocation = http.getHeaderField("Location");
@@ -216,12 +231,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);
 		}
@@ -272,7 +291,7 @@ public String[] list(final String normalPath) throws N5IOException {
 
 	private String[] queryListEntries(String normalPath, ListResponseParser parser, boolean allowRedirect) throws N5IOException{
 
-		final HttpURLConnection http = requireValidHttpResponse(normalPath, "GET", "Error listing directory at " + normalPath, allowRedirect);
+		final HttpURLConnection http = requireValidHttpResponse(normalPath, GET, "Error listing directory at " + normalPath, allowRedirect);
 		try {
 			final String listResponse = responseToString(http.getInputStream());
 			return parser.parseListResponse(listResponse);
@@ -333,23 +352,47 @@ 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 (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 {
 
@@ -384,4 +427,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);
+			}
+		}
+	}
+
 }