Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
e74fd91
fix: AbstractInputStreamReadData close stream
bogovicj May 20, 2025
226e493
feat: add split and limit methods to ReadData
bogovicj Jun 2, 2025
61dc420
feat: add KeyValueAccessLazyReadData
bogovicj Jun 3, 2025
83b1ea8
refactor: ByteArrayReadData (rm "Splittable")
bogovicj Jun 3, 2025
e276756
doc: javadoc fixes
bogovicj Jun 3, 2025
7df5b07
refactor/doc: rename KeyValueAccessLazyReadData.lazySlice
bogovicj Jun 3, 2025
8170919
perf/test: slice throws exception immediately when possible
bogovicj Jun 12, 2025
05280a8
feat: add ReadData.empty()
bogovicj Jun 12, 2025
52d86d9
remove ReadData.split
bogovicj Jun 12, 2025
d5d309c
style: rm unused imports
bogovicj Jun 13, 2025
50ae586
Merge remote-tracking branch 'upstream/readDataN5IOExceptions' into s…
bogovicj Jun 13, 2025
198e0d5
Merge remote-tracking branch 'upstream/master' into splittable-readdata
bogovicj Jun 13, 2025
725c0b2
doc: javadoc fixes
bogovicj Jun 13, 2025
20c6de2
style: rm unused import
bogovicj Jun 16, 2025
9e8aa91
test/style: rm unused pivot test argument
bogovicj Jun 16, 2025
c4c93b5
test/style: make ByteFun test class static
bogovicj Jun 16, 2025
bd5d072
test: slice testing works on unmaterialized ReadData
bogovicj Jun 16, 2025
493276e
style!: remove ReadData.from(KeyValueAccess)
bogovicj Jun 16, 2025
a90db72
fix: KVA.createReadData is now abstract
bogovicj Jun 16, 2025
8b69fa7
refactor!: implementation of KVA.createReadData
bogovicj Jun 17, 2025
e1b9cf9
refactor: move LazyRead into KeyValueAccess
bogovicj Jun 17, 2025
ca470ab
style: rm redundant modifiers
tpietzsch Jun 22, 2025
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -85,6 +86,8 @@
import java.util.Iterator;
import java.util.stream.Stream;

import org.janelia.saalfeldlab.n5.readdata.ReadData;

/**
* Filesystem {@link KeyValueAccess}.
*
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 {

Expand Down Expand Up @@ -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;
}

}
80 changes: 74 additions & 6 deletions src/main/java/org/janelia/saalfeldlab/n5/HttpKeyValueAccess.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -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>
Expand All @@ -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");
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 {

Expand Down Expand Up @@ -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);
}
}
}

}
Loading