diff --git a/pom.xml b/pom.xml index 98f268f4c..b4094436b 100644 --- a/pom.xml +++ b/pom.xml @@ -202,10 +202,6 @@ org.apache.commons commons-compress - - commons-io - commons-io - diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java index 4aeafe8bb..d6a9803b8 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java +++ b/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; + } + } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/HttpKeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/HttpKeyValueAccess.java index d21505cd2..8826ebaf4 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/HttpKeyValueAccess.java +++ b/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. *

@@ -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 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); + } + } + } + } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccess.java index 5ae2d23c3..5ed6eec8a 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccess.java @@ -61,6 +61,8 @@ import java.util.Arrays; import java.util.stream.Collectors; +import org.janelia.saalfeldlab.n5.readdata.ReadData; + /** * Key value read primitives used by {@link N5KeyValueReader} * implementations. This interface implements a subset of access primitives @@ -79,7 +81,7 @@ public interface KeyValueAccess { * the path * @return the path components */ - public default String[] components(final String path) { + default String[] components( final String path ) { String[] components = Arrays.stream(path.split("/")) .filter(x -> !x.isEmpty()) @@ -107,7 +109,7 @@ public default String[] components(final String path) { * @param components the path components * @return the path */ - public default String compose(final URI uri, final String... components) { + default String compose( final URI uri, final String... components ) { int firstNonEmptyIdx = 0; while (firstNonEmptyIdx < components.length && (components[firstNonEmptyIdx] == null || components[firstNonEmptyIdx].isEmpty())) { @@ -146,7 +148,7 @@ else if (component.endsWith("/") || i == allComponents.length - 1) } @Deprecated - public default String compose(final String... components) { + default String compose( final String... components ) { return normalize( Arrays.stream(components) @@ -163,7 +165,7 @@ public default String compose(final String... components) { * the path * @return the parent path or null if the path has no parent */ - public default String parent(final String path) { + default String parent( final String path ) { final String removeTrailingSlash = path.replaceAll("/+$", ""); return normalize(N5URI.getAsUri(removeTrailingSlash).resolve("").toString()); } @@ -177,7 +179,7 @@ public default String parent(final String path) { * the base path * @return the result or null if the path has no parent */ - public default String relativize(final String path, final String base) { + default String relativize( final String path, final String base ) { try { /* @@ -203,7 +205,7 @@ public default String relativize(final String path, final String base) { * the path * @return the normalized path */ - public String normalize(final String path); + String normalize( final String path ); /** * Get the absolute (including scheme) {@link URI} of the given path @@ -229,7 +231,18 @@ default URI uri(final String uriString) throws URISyntaxException { * efforts are made to normalize it. * @return true if the path exists */ - public boolean exists(final String normalPath); + boolean exists( final String normalPath ); + + /** + * Returns the size in bytes of the object at the given normalPath if it exists. + * + * @param normalPath + * is expected to be in normalized form, no further + * efforts are made to normalize it. + * @return the size of the object in bytes. + * @throws N5Exception.N5NoSuchKeyException if the given key does not exist + */ + long size( final String normalPath ) throws N5Exception.N5NoSuchKeyException; /** * Test whether the path is a directory. @@ -239,7 +252,7 @@ default URI uri(final String uriString) throws URISyntaxException { * efforts are made to normalize it. * @return true if the path is a directory */ - public boolean isDirectory(String normalPath); + boolean isDirectory( String normalPath ); /** * Test whether the path is a file. @@ -249,7 +262,21 @@ default URI uri(final String uriString) throws URISyntaxException { * efforts are made to normalize it. * @return true if the path is a file */ - public boolean isFile(String normalPath); // TODO: Looks un-used. Remove? + boolean isFile( String normalPath ); // TODO: Looks un-used. Remove? + + /** + * Create a {@link ReadData} through which data at the normal key can be read. + *

+ * Implementations should read lazily if possible. Consumers may call {@link ReadData#materialize()} to force + * a read operation if needed. + *

+ * If supported by this KeyValueAccess implementation, partial reads are possible by calling slice on the output {@link ReadData}. + * + * @param normalPath is expected to be in normalized form, no further efforts are made to normalize it + * @return a materialized Read data + * @throws N5IOException if an error occurs + */ + ReadData createReadData( final String normalPath ) throws N5IOException; /** * Create a lock on a path for reading. This isn't meant to be kept @@ -267,7 +294,7 @@ default URI uri(final String uriString) throws URISyntaxException { * @throws N5IOException * if a locked channel could not be created */ - public LockedChannel lockForReading(final String normalPath) throws N5IOException; + LockedChannel lockForReading( final String normalPath ) throws N5IOException; /** * Create an exclusive lock on a path for writing. If the file doesn't @@ -287,7 +314,7 @@ default URI uri(final String uriString) throws URISyntaxException { * @throws N5IOException * if a locked channel could not be created */ - public LockedChannel lockForWriting(final String normalPath) throws N5IOException; + LockedChannel lockForWriting( final String normalPath ) throws N5IOException; /** * List all 'directory'-like children of a path. @@ -299,7 +326,7 @@ default URI uri(final String uriString) throws URISyntaxException { * @throws N5IOException * if an error occurs during listing */ - public String[] listDirectories(final String normalPath) throws N5IOException; + String[] listDirectories( final String normalPath ) throws N5IOException; /** * List all children of a path. @@ -310,7 +337,7 @@ default URI uri(final String uriString) throws URISyntaxException { * @return the the child paths * @throws N5IOException if an error occurs during listing */ - public String[] list(final String normalPath) throws N5IOException; + String[] list( final String normalPath ) throws N5IOException; /** * Create a directory and all parent paths along the way. The directory @@ -324,7 +351,7 @@ default URI uri(final String uriString) throws URISyntaxException { * @throws N5IOException * if an error occurs during creation */ - public void createDirectories(final String normalPath) throws N5IOException; + void createDirectories( final String normalPath ) throws N5IOException; /** * Delete a path. If the path is a directory, delete it recursively. @@ -335,5 +362,48 @@ default URI uri(final String uriString) throws URISyntaxException { * @throws N5IOException * if an error occurs during deletion */ - public void delete(final String normalPath) throws N5IOException; + void delete( final String normalPath ) throws N5IOException; + + /** + * A lazy reading strategy for lazy, partial reading of data from some source. + *

+ * Implementations of this interface handle the specifics of accessing data from + * their respective sources. + * + * @see ReadData + * @see KeyValueAccessReadData + */ + interface LazyRead { + + /** + * Materializes a portion of the data into a concrete {@link ReadData} + * instance. + *

+ * This method performs the actual read operation from the underlying + * source, loading only the requested portion of data. The implementation + * should handle bounds checking and throw appropriate exceptions for + * invalid ranges. + * + * @param offset + * the starting position in the data source + * @param length + * the number of bytes to read, or -1 to read from offset to end + * @return a materialized {@link ReadData} instance containing the requested + * data + * @throws N5IOException + * if any I/O error occurs + */ + ReadData materialize(long offset, long length) throws N5IOException; + + /** + * Returns the total size of the data source in bytes. + * + * @return the size of the data source in bytes + * @throws N5IOException + * if an I/O error occurs while trying to get the length + */ + long size() throws N5IOException; + + } + } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccessReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccessReadData.java new file mode 100644 index 000000000..77c3c4efd --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccessReadData.java @@ -0,0 +1,88 @@ +package org.janelia.saalfeldlab.n5; + +import java.io.InputStream; + +import org.janelia.saalfeldlab.n5.KeyValueAccess.LazyRead; +import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; +import org.janelia.saalfeldlab.n5.readdata.ReadData; + +/** + * A {@link ReadData} implementation that reads from a {@link KeyValueAccess} + * backend through a {@link LazyRead} object. + */ +public class KeyValueAccessReadData implements ReadData { + + private final LazyRead lazyRead; + private ReadData materialized; + private final long offset; + private long length; + + KeyValueAccessReadData(LazyRead lazyRead) { + this(lazyRead, 0, -1); + } + + KeyValueAccessReadData(final LazyRead lazyRead, final long offset, final long length) { + this.lazyRead = lazyRead; + this.offset = offset; + this.length = length; + } + + @Override + public ReadData materialize() throws N5IOException { + if (materialized == null) + materialized = lazyRead.materialize(offset, length); + return materialized; + } + + /** + * Returns a {@link ReadData} whose length is limited to the given value. + *

+ * This implementation defers a material read operation if allowed + * by the {@link LazyRead}. + * + * @param length + * the length of the resulting ReadData + * @return a length-limited ReadData + * @throws N5IOException + * if an I/O error occurs while trying to get the length + */ + @Override + 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; + if (this.length > 0 && length < 0) + lengthArg = (int)(this.length - offset); + else + lengthArg = (int)length; + + return new KeyValueAccessReadData(lazyRead, this.offset + offset, lengthArg); + } + + @Override + public InputStream inputStream() throws N5IOException, IllegalStateException { + return materialize().inputStream(); + } + + @Override + public byte[] allBytes() throws N5IOException, IllegalStateException { + return materialize().allBytes(); + } + + @Override + public long length() throws N5IOException { + if (materialized != null) + return materialized.length(); + if (length < 0) { + length = lazyRead.size() - offset; + } + return length; + } + +} diff --git a/src/main/java/org/janelia/saalfeldlab/n5/readdata/AbstractInputStreamReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/readdata/AbstractInputStreamReadData.java index 0034a2e75..854f7d4f5 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/readdata/AbstractInputStreamReadData.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/readdata/AbstractInputStreamReadData.java @@ -38,7 +38,7 @@ // not thread-safe abstract class AbstractInputStreamReadData implements ReadData { - private ByteArraySplittableReadData bytes; + private ByteArrayReadData bytes; @Override public ReadData materialize() throws N5IOException { @@ -59,7 +59,7 @@ public ReadData materialize() throws N5IOException { throw new N5IOException(e); } } - bytes = new ByteArraySplittableReadData(data); + bytes = new ByteArrayReadData(data); } return bytes; } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/readdata/ByteArraySplittableReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/readdata/ByteArrayReadData.java similarity index 68% rename from src/main/java/org/janelia/saalfeldlab/n5/readdata/ByteArraySplittableReadData.java rename to src/main/java/org/janelia/saalfeldlab/n5/readdata/ByteArrayReadData.java index 3077f4ab4..92dcbd820 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/readdata/ByteArraySplittableReadData.java +++ b/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 @@ -32,20 +32,32 @@ import java.io.InputStream; import java.util.Arrays; -class ByteArraySplittableReadData implements ReadData { +class ByteArrayReadData implements ReadData { + + static final ReadData EMPTY = new ByteArrayReadData(new byte[0]); private final byte[] data; private final int offset; private final int length; - ByteArraySplittableReadData(final byte[] data) { + ByteArrayReadData(final byte[] data) { + this(data, 0, data.length); } - ByteArraySplittableReadData(final byte[] data, final int offset, final int length) { + ByteArrayReadData(final byte[] data, final int offset, final int length) { + + if (!validBounds(data.length, offset, length)) + throw new IndexOutOfBoundsException(); + this.data = data; this.offset = offset; - this.length = length; + + if( length < 0 ) + this.length = data.length - offset; + else + this.length = length; + } @Override @@ -60,6 +72,8 @@ public InputStream inputStream() { @Override public byte[] allBytes() { + + // alternatively, we could always return the requested length if (offset == 0 && data.length == length) { return data; } else { @@ -71,4 +85,24 @@ public byte[] allBytes() { public ReadData materialize() { return this; } + + @Override + public ReadData slice(final long offset, final long length) { + + final int o = this.offset + (int)offset; + return new ByteArrayReadData(data, o, (int)length); + } + + private static boolean validBounds(int arrayLength, int offset, int length) { + + if (offset < 0) + return false; + else if (arrayLength > 0 && offset >= arrayLength) // offset == 0 and arrayLength == 0 is okay + return false; + else if (length >= 0 && offset + length > arrayLength) + return false; + + return true; + } + } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/readdata/KeyValueAccessReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/readdata/KeyValueAccessReadData.java deleted file mode 100644 index 2fa2ca00d..000000000 --- a/src/main/java/org/janelia/saalfeldlab/n5/readdata/KeyValueAccessReadData.java +++ /dev/null @@ -1,74 +0,0 @@ -/*- - * #%L - * Not HDF5 - * %% - * Copyright (C) 2017 - 2025 Stephan Saalfeld - * %% - * 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 - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - * #L% - */ -package org.janelia.saalfeldlab.n5.readdata; - -import java.io.IOException; -import java.io.InputStream; -import org.apache.commons.io.input.ProxyInputStream; -import org.janelia.saalfeldlab.n5.KeyValueAccess; -import org.janelia.saalfeldlab.n5.LockedChannel; -import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; - -class KeyValueAccessReadData extends AbstractInputStreamReadData { - - private final KeyValueAccess keyValueAccess; - private final String normalPath; - - KeyValueAccessReadData(final KeyValueAccess keyValueAccess, final String normalPath) { - this.keyValueAccess = keyValueAccess; - this.normalPath = normalPath; - } - - /** - * Open a {@code InputStream} on this data. - *

- * This will open a {@code LockedChannel} on the underlying {@code - * KeyValueAccess}. Make sure to {@code close()} the returned {@code - * InputStream} to release the underlying {@code LockedChannel}. - * - * @return an InputStream on this data - * - * @throws N5IOException - * if any I/O error occurs - */ - @Override - public InputStream inputStream() throws N5IOException { - - @SuppressWarnings("resource") - final LockedChannel channel = keyValueAccess.lockForReading(normalPath); - return new ProxyInputStream(channel.newInputStream()) { - - @Override - public void close() throws IOException { - in.close(); - channel.close(); - } - }; - } -} diff --git a/src/main/java/org/janelia/saalfeldlab/n5/readdata/LazyReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/readdata/LazyReadData.java index 40ee709a7..9134e0ba2 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/readdata/LazyReadData.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/readdata/LazyReadData.java @@ -33,7 +33,6 @@ import java.io.InputStream; import java.io.OutputStream; import org.apache.commons.io.output.ProxyOutputStream; -import org.janelia.saalfeldlab.n5.N5Exception; import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; class LazyReadData implements ReadData { @@ -61,14 +60,14 @@ class LazyReadData implements ReadData { private final OutputStreamWriter writer; - private ByteArraySplittableReadData bytes; + private ByteArrayReadData bytes; @Override public ReadData materialize() throws N5IOException { if (bytes == null) { final ByteArrayOutputStream baos = new ByteArrayOutputStream(8192); writeTo(baos); - bytes = new ByteArraySplittableReadData(baos.toByteArray()); + bytes = new ByteArrayReadData(baos.toByteArray()); } return bytes; } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/readdata/ReadData.java b/src/main/java/org/janelia/saalfeldlab/n5/readdata/ReadData.java index 71f3c1325..724e71cdc 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/readdata/ReadData.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/readdata/ReadData.java @@ -32,6 +32,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; + import org.janelia.saalfeldlab.n5.KeyValueAccess; import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; @@ -67,6 +68,32 @@ default long length() throws N5IOException { return -1; } + /** + * Returns a {@link ReadData} whose length is limited to the given value. + * + * @param length + * the length of the resulting ReadData + * @return a length-limited ReadData + * @throws N5IOException + * if an I/O error occurs while trying to get the length + */ + default ReadData limit(final long length) throws N5IOException { + return slice(0, length); + } + + /** + * Returns a new {@link ReadData} representing a slice, or subset + * of this ReadData. + * + * @param offset the offset relative to this + * @param length of the returned ReadData + * @return a slice + * @throws N5IOException an exception + */ + default ReadData slice(final long offset, final long length) throws N5IOException { + return materialize().slice(offset, length); + } + /** * Open a {@code InputStream} on this data. *

@@ -133,6 +160,11 @@ default ByteBuffer toByteBuffer() throws N5IOException, IllegalStateException { *

* The returned {@code ReadData} has a known {@link #length} and multiple * {@link #inputStream InputStreams} can be opened on it. + * + * @return + * a materialized ReadData. + * @throws N5IOException + * if any I/O error occurs */ ReadData materialize() throws N5IOException; @@ -222,22 +254,6 @@ static ReadData from(final InputStream inputStream) { return from(inputStream, -1); } - /** - * Create a new {@code ReadData} that loads lazily from {@code normalPath} - * in {@code keyValueAccess}. The returned ReadData reports {@link #length() - * length() == -1} (i.e., unknown length). - * - * @param keyValueAccess - * KeyValueAccess to read from - * @param normalPath - * path in the {@code keyValueAccess} to read from - * - * @return a new ReadData - */ - static ReadData from(final KeyValueAccess keyValueAccess, final String normalPath) { - return new KeyValueAccessReadData(keyValueAccess, normalPath); - } - /** * Create a new {@code ReadData} that wraps the specified portion of a * {@code byte[]} array. @@ -252,7 +268,7 @@ static ReadData from(final KeyValueAccess keyValueAccess, final String normalPat * @return a new ReadData */ static ReadData from(final byte[] data, final int offset, final int length) { - return new ByteArraySplittableReadData(data, offset, length); + return new ByteArrayReadData(data, offset, length); } /** @@ -299,4 +315,14 @@ interface OutputStreamWriter { static ReadData from(OutputStreamWriter generator) { return new LazyReadData(generator); } + + /** + * Returns an empty {@code ReadData}. + * + * @return an empty ReadData + */ + public static ReadData empty() { + return ByteArrayReadData.EMPTY; + } + } diff --git a/src/test/java/org/janelia/saalfeldlab/n5/readdata/ReadDataTests.java b/src/test/java/org/janelia/saalfeldlab/n5/readdata/ReadDataTests.java new file mode 100644 index 000000000..b2e7f3c27 --- /dev/null +++ b/src/test/java/org/janelia/saalfeldlab/n5/readdata/ReadDataTests.java @@ -0,0 +1,156 @@ +package org.janelia.saalfeldlab.n5.readdata; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.FileSystems; +import java.util.Arrays; +import java.util.function.IntUnaryOperator; + +import org.apache.commons.compress.utils.IOUtils; +import org.janelia.saalfeldlab.n5.FileSystemKeyValueAccess; +import org.janelia.saalfeldlab.n5.readdata.ReadData.OutputStreamOperator; +import org.junit.Test; + +public class ReadDataTests { + + @Test + public void testLazyReadData() throws IOException { + + final int N = 128; + byte[] data = new byte[N]; + for( int i = 0; i < N; i++ ) + data[i] = (byte)i; + + final ReadData readData = ReadData.from(out -> { + out.write(data); + }); + assertTrue(readData instanceof LazyReadData); + + readDataTestHelper(readData, N); + sliceTestHelper(readData, N); + } + + @Test + public void testByteArrayReadData() throws IOException { + + final int N = 128; + byte[] data = new byte[N]; + for( int i = 0; i < N; i++ ) + data[i] = (byte)i; + + ReadData readData = ReadData.from(data).materialize(); + assertTrue(readData instanceof ByteArrayReadData); + + readDataTestHelper(readData, N); + readDataTestEncodeHelper(readData, N); + sliceTestHelper(readData, N); + } + + @Test + public void testInputStreamReadData() throws IOException { + + final int N = 128; + final InputStream is = new InputStream() { + int val = 0; + @Override + public int read() throws IOException { + return val++; + } + }; + + final ReadData readData = ReadData.from(is, N); + readDataTestHelper(readData, N); + sliceTestHelper(readData, N); + } + + @Test + public void testFileKvaReadData() throws IOException { + + int N = 128; + byte[] data = new byte[N]; + for( int i = 0; i < N; i++ ) + data[i] = (byte)i; + + final File tmpF = File.createTempFile("test-file-splittable-data", ".bin"); + tmpF.deleteOnExit(); + try (FileOutputStream os = new FileOutputStream(tmpF)) { + os.write(data); + } + + final ReadData readData = new FileSystemKeyValueAccess(FileSystems.getDefault()) + .createReadData(tmpF.getAbsolutePath()); + + assertEquals("file read data length", 128, readData.length()); + sliceTestHelper(readData, N); + } + + private void readDataTestHelper( ReadData readData, int N ) throws IOException { + + assertEquals("full length", N, readData.length()); + } + + private void readDataTestEncodeHelper( ReadData readData, int N ) throws IOException { + + final byte[] origCopy = new byte[N]; + IOUtils.readFully(readData.inputStream(), origCopy); + + final byte[] expected = Arrays.copyOf(origCopy, N); + for( int i = 0; i < expected.length; i++) + expected[i]+=2; + + final ReadData encoded = readData.encode(new ByteFun(x -> x+2)); + assertArrayEquals(expected, encoded.allBytes()); + + final ReadData encodedTwice = encoded.encode(new ByteFun(x -> x-2)); + assertArrayEquals(origCopy, encodedTwice.allBytes()); + } + + private void sliceTestHelper( ReadData readData, int N ) throws IOException { + + assertEquals("length one", 1, readData.slice(9, 1).length()); + + assertEquals("split length zero", 0, readData.slice(9, 0).length()); + assertEquals("split length zero allBytes", 0, readData.slice(9, 0).allBytes().length); + + ReadData limited = readData.limit(2); + assertEquals(2, limited.length()); + + ReadData unboundedLength = readData.slice(1, -1); + assertEquals("unbounded length allBytes", N - 1, unboundedLength.allBytes().length); + + // slice may throw an exception if it knows its length and can detect out-of-bounds + // otherwise the exception may be thrown on a read operation (e.g. allBytes) + assertThrows("Out-of-range slice read", IndexOutOfBoundsException.class, () -> readData.slice(N-1, 3).allBytes()); + assertThrows("slice throws if offset too large", IndexOutOfBoundsException.class, () -> readData.slice(N, 0).allBytes()); + assertThrows("too large offset slice read", IndexOutOfBoundsException.class, () -> readData.slice(N-1, 3).allBytes()); + + assertThrows("negative offset", IndexOutOfBoundsException.class, () -> readData.slice(-1, 1)); + } + + private static class ByteFun implements OutputStreamOperator { + + IntUnaryOperator fun; + public ByteFun(IntUnaryOperator fun) { + this.fun = fun; + } + + @Override + public OutputStream apply(OutputStream o) throws IOException { + return new OutputStream() { + @Override + public void write(int b) throws IOException { + o.write(fun.applyAsInt(b)); + } + }; + } + } + +}