add store.getInputStream

brokkoli71 · brokkoli71 · commit 5c74445f353b · 2025-12-18T15:47:13.000+01:00
diff --git a/src/main/java/dev/zarr/zarrjava/store/BufferedZipStore.java b/src/main/java/dev/zarr/zarrjava/store/BufferedZipStore.java
@@ -4,6 +4,7 @@
 import javax.annotation.Nullable;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -290,6 +291,11 @@ public StoreHandle resolve(String... keys) {
         return new StoreHandle(this, keys);
     }
 
+    @Override
+    public InputStream getInputStream(String[] keys, long start, long end) {
+        return bufferStore.getInputStream(keys, start, end);
+    }
+
     @Override
     public String toString() {
         return "BufferedZipStore(" + underlyingStore.toString() + ")";
diff --git a/src/main/java/dev/zarr/zarrjava/store/FilesystemStore.java b/src/main/java/dev/zarr/zarrjava/store/FilesystemStore.java
@@ -1,7 +1,10 @@
 package dev.zarr.zarrjava.store;
 
 import dev.zarr.zarrjava.utils.Utils;
+import org.apache.commons.io.input.BoundedInputStream;
+
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.SeekableByteChannel;
 import java.nio.file.Files;
@@ -146,4 +149,25 @@ public String toString() {
     return this.path.toUri().toString().replaceAll("\\/$", "");
   }
 
+    @Override
+    public InputStream getInputStream(String[] keys, long start, long end) {
+        Path keyPath = resolveKeys(keys);
+        try {
+            InputStream inputStream = Files.newInputStream(keyPath);
+            if (start > 0) {
+                long skipped = inputStream.skip(start);
+                if (skipped < start) {
+                    throw new IOException("Unable to skip to the desired start position.");
+                }
+            }
+            if (end != -1) {
+                long bytesToRead = end - start;
+                return new BoundedInputStream(inputStream, bytesToRead);
+            } else {
+                return inputStream;
+            }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }
diff --git a/src/main/java/dev/zarr/zarrjava/store/HttpStore.java b/src/main/java/dev/zarr/zarrjava/store/HttpStore.java
@@ -5,7 +5,10 @@
 import com.squareup.okhttp.Request;
 import com.squareup.okhttp.Response;
 import com.squareup.okhttp.ResponseBody;
+
+import java.io.FilterInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -101,6 +104,34 @@ public StoreHandle resolve(String... keys) {
 
   @Override
   public String toString() {
-    return uri;
+      return uri;
   }
+
+    @Override
+    @Nullable
+    public InputStream getInputStream(String[] keys, long start, long end) {
+        if (start < 0) {
+            throw new IllegalArgumentException("Argument 'start' needs to be non-negative.");
+        }
+        Request request = new Request.Builder().url(resolveKeys(keys)).header(
+                "Range", String.format("Bytes=%d-%d", start, end - 1)).build();
+        Call call = httpClient.newCall(request);
+        try {
+            Response response = call.execute();
+            ResponseBody body = response.body();
+            if (body == null) return null;
+            InputStream stream = body.byteStream();
+
+            // Ensure closing the stream also closes the response
+            return new FilterInputStream(stream) {
+                @Override
+                public void close() throws IOException {
+                    super.close();
+                    body.close();
+                }
+            };
+        } catch (IOException e) {
+            return null;
+        }
+    }
 }
diff --git a/src/main/java/dev/zarr/zarrjava/store/MemoryStore.java b/src/main/java/dev/zarr/zarrjava/store/MemoryStore.java
@@ -2,6 +2,7 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -45,7 +46,7 @@ public ByteBuffer get(String[] keys, long start, long end) {
     if (bytes == null) return null;
     if (end < 0) end = bytes.length;
     if (end > Integer.MAX_VALUE) throw new IllegalArgumentException("End index too large");
-    return ByteBuffer.wrap(bytes, (int) start, (int) end);
+    return ByteBuffer.wrap(bytes, (int) start, (int) (end - start));
   }
 
 
@@ -83,5 +84,14 @@ public StoreHandle resolve(String... keys) {
   public String toString() {
     return String.format("<MemoryStore {%s}>", hashCode());
   }
+
+    @Override
+    public InputStream getInputStream(String[] keys, long start, long end) {
+        byte[] bytes = map.get(resolveKeys(keys));
+        if (bytes == null) return null;
+        if (end < 0) end = bytes.length;
+        if (end > Integer.MAX_VALUE) throw new IllegalArgumentException("End index too large");
+        return new java.io.ByteArrayInputStream(bytes, (int) start, (int)(end - start));
+    }
 }
 
diff --git a/src/main/java/dev/zarr/zarrjava/store/ReadOnlyZipStore.java b/src/main/java/dev/zarr/zarrjava/store/ReadOnlyZipStore.java
@@ -3,11 +3,13 @@
 import com.fasterxml.jackson.databind.util.ByteBufferBackedInputStream;
 import org.apache.commons.compress.archivers.zip.ZipArchiveEntry;
 import org.apache.commons.compress.archivers.zip.ZipArchiveInputStream;
+import org.apache.commons.io.input.BoundedInputStream;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -83,7 +85,8 @@ public ByteBuffer get(String[] keys, long start, long end) {
                     continue;
                 }
 
-                if (zis.skip(start) != start) {
+                long skipResult = zis.skip(start);
+                if (skipResult != start) {
                     throw new IOException("Failed to skip to start position " + start + " in zip entry " + entryName);
                 }
 
@@ -168,4 +171,38 @@ public Stream<String[]> list(String[] keys) {
         }
         return builder.build();
     }
+
+    @Override
+    public InputStream getInputStream(String[] keys, long start, long end) {
+        InputStream baseStream = underlyingStore.getInputStream();
+
+        try {
+            ZipArchiveInputStream zis = new ZipArchiveInputStream(baseStream);
+            ZipArchiveEntry entry;
+            while ((entry = zis.getNextEntry()) != null) {
+                String entryName = entry.getName();
+
+                if (entryName.startsWith("/")) {
+                    entryName = entryName.substring(1);
+                }
+                if (entry.isDirectory() || !entryName.equals(resolveKeys(keys))) {
+                    continue;
+                }
+
+                long skipResult = zis.skip(start);
+                if (skipResult != start) {
+                    throw new IOException("Failed to skip to start position " + start + " in zip entry " + entryName);
+                }
+
+                long bytesToRead;
+                if (end != -1) bytesToRead = end - start;
+                else bytesToRead = Long.MAX_VALUE;
+
+                return new BoundedInputStream(zis, bytesToRead);
+            }
+            return null;
+        } catch (IOException e) {
+        }
+        return null;
+    }
 }
diff --git a/src/main/java/dev/zarr/zarrjava/store/S3Store.java b/src/main/java/dev/zarr/zarrjava/store/S3Store.java
@@ -121,6 +121,17 @@ public StoreHandle resolve(String... keys) {
     return new StoreHandle(this, keys);
   }
 
+  @Override
+  public InputStream getInputStream(String[] keys, long start, long end) {
+    GetObjectRequest req = GetObjectRequest.builder()
+    .bucket(bucketName)
+    .key(resolveKeys(keys))
+    .range(String.format("bytes=%d-%d", start, end-1)) // S3 range is inclusive
+    .build();
+    ResponseInputStream<GetObjectResponse> responseInputStream = s3client.getObject(req);
+    return responseInputStream;
+  }
+
   @Override
   public String toString() {
     return "s3://" + bucketName + "/" + prefix;
diff --git a/src/main/java/dev/zarr/zarrjava/store/Store.java b/src/main/java/dev/zarr/zarrjava/store/Store.java
@@ -1,5 +1,6 @@
 package dev.zarr.zarrjava.store;
 
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.stream.Stream;
 import javax.annotation.Nonnull;
@@ -42,4 +43,10 @@ default Stream<String[]> list() {
         return list(new String[]{});
     }
   }
+
+  InputStream getInputStream(String[] keys, long start, long end);
+
+  default InputStream getInputStream(String[] keys) {
+    return getInputStream(keys, 0, -1);
+  }
 }
diff --git a/src/main/java/dev/zarr/zarrjava/store/StoreHandle.java b/src/main/java/dev/zarr/zarrjava/store/StoreHandle.java
@@ -1,6 +1,8 @@
 package dev.zarr.zarrjava.store;
 
 import dev.zarr.zarrjava.utils.Utils;
+
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
@@ -44,6 +46,14 @@ public ByteBuffer read(long start, long end) {
     return store.get(keys, start, end);
   }
 
+  public InputStream getInputStream(int start, int end) {
+    return store.getInputStream(keys, start, end);
+  }
+
+  public InputStream getInputStream() {
+    return store.getInputStream(keys);
+  }
+
   public void set(ByteBuffer bytes) {
     store.set(keys, bytes);
   }
diff --git a/src/test/java/dev/zarr/zarrjava/ZarrStoreTest.java b/src/test/java/dev/zarr/zarrjava/ZarrStoreTest.java
@@ -18,6 +18,8 @@
 import software.amazon.awssdk.services.s3.S3Client;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -103,6 +105,55 @@ public void testS3StoreGet() throws IOException, ZarrException {
 
     }
 
+    static Stream<StoreHandle> inputStreamStores() throws IOException {
+        String[] s3StoreKeys = new String[]{"zarr_v3", "l4_sample", "color", "1", "zarr.json"};
+        StoreHandle s3StoreHandle = new S3Store(S3Client.builder()
+                .region(Region.of("eu-west-1"))
+                .credentialsProvider(AnonymousCredentialsProvider.create())
+                .build(), "static.webknossos.org", "data")
+                .resolve(s3StoreKeys);
+
+        byte[] testData = new byte[100];
+        for (int i = 0; i < testData.length; i++) {
+            testData[i] = (byte) i;
+        }
+
+        StoreHandle memoryStoreHandle = new MemoryStore().resolve();
+        memoryStoreHandle.set(ByteBuffer.wrap(testData));
+
+        StoreHandle fsStoreHandle = new FilesystemStore(TESTOUTPUT.resolve("testInputStreamFS")).resolve("testfile");
+        fsStoreHandle.set(ByteBuffer.wrap(testData));
+
+        zipFile(TESTOUTPUT.resolve("testInputStreamFS"), TESTOUTPUT.resolve("testInputStreamZIP.zip"));
+        StoreHandle bufferedZipStoreHandle = new BufferedZipStore(TESTOUTPUT.resolve("testInputStreamZIP.zip"), true)
+                .resolve("testfile");
+
+        StoreHandle readOnlyZipStoreHandle = new ReadOnlyZipStore(TESTOUTPUT.resolve("testInputStreamZIP.zip"))
+                .resolve("testfile");
+
+        StoreHandle httpStoreHandle = new HttpStore("https://static.webknossos.org/data/zarr_v3/l4_sample")
+                .resolve("color", "1", "zarr.json");
+        return Stream.of(
+                memoryStoreHandle,
+                s3StoreHandle,
+                fsStoreHandle,
+                bufferedZipStoreHandle,
+                readOnlyZipStoreHandle,
+                httpStoreHandle
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("inputStreamStores")
+    public void testStoreInputStream(StoreHandle storeHandle) throws IOException, ZarrException {
+        InputStream is = storeHandle.getInputStream(10, 20);
+        byte[] buffer = new byte[10];
+        int bytesRead = is.read(buffer);
+        Assertions.assertEquals(10, bytesRead);
+        byte[] expectedBuffer = new byte[10];
+        storeHandle.read(10, 20).get(expectedBuffer);
+        Assertions.assertArrayEquals(expectedBuffer, buffer);
+    }
 
     @Test
     public void testHttpStore() throws IOException, ZarrException {
@@ -115,8 +166,7 @@ public void testHttpStore() throws IOException, ZarrException {
     @ParameterizedTest
     @CsvSource({"false", "true",})
     public void testMemoryStoreV3(boolean useParallel) throws ZarrException, IOException {
-        int[] testData = new int[1024 * 1024];
-        Arrays.setAll(testData, p -> p);
+        int[] testData = testData();
 
         dev.zarr.zarrjava.v3.Group group = dev.zarr.zarrjava.v3.Group.create(new MemoryStore().resolve());
         Array array = group.createArray("array", b -> b
@@ -140,8 +190,7 @@ public void testMemoryStoreV3(boolean useParallel) throws ZarrException, IOExcep
     @ParameterizedTest
     @CsvSource({"false", "true",})
     public void testMemoryStoreV2(boolean useParallel) throws ZarrException, IOException {
-        int[] testData = new int[1024 * 1024];
-        Arrays.setAll(testData, p -> p);
+        int[] testData = testData();
 
         dev.zarr.zarrjava.v2.Group group = dev.zarr.zarrjava.v2.Group.create(new MemoryStore().resolve());
         dev.zarr.zarrjava.v2.Array array = group.createArray("array", b -> b

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`package dev.zarr.zarrjava.store;`
`2`	`2`
	`3`	`+import java.io.InputStream;`
`3`	`4`	`import java.nio.ByteBuffer;`
`4`	`5`	`import java.util.stream.Stream;`
`5`	`6`	`import javax.annotation.Nonnull;`
`@@ -42,4 +43,10 @@ default Stream<String[]> list() {`
`42`	`43`	`return list(new String[]{});`
`43`	`44`	`}`
`44`	`45`	`}`
	`46`	`+`
	`47`	`+ InputStream getInputStream(String[] keys, long start, long end);`
	`48`	`+`
	`49`	`+ default InputStream getInputStream(String[] keys) {`
	`50`	`+ return getInputStream(keys, 0, -1);`
	`51`	`+ }`
`45`	`52`	`}`