Fixing streaming of files and generalizing for other storage providers.

Author: luis100

roda-core/roda-core/src/main/java/org/roda/core/storage/RangeConsumesOutputStream.java

@@ -7,88 +7,78 @@
  */
 package org.roda.core.storage;
 
-import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.OutputStream;
-import java.io.RandomAccessFile;
-import java.nio.file.Files;
-import java.nio.file.Path;
 import java.util.Date;
 
-import org.apache.commons.io.IOUtils;
 import org.roda.core.data.v2.ConsumesSkipableOutputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class RangeConsumesOutputStream implements ConsumesSkipableOutputStream {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(RangeConsumesOutputStream.class);
+
   private static final String DEFAULT_MIME_TYPE = "application/octet-stream";
-  private final Path directAccessPath;
+  private final SeekableContentPayload payload;
+  private final String filename;
   private final String mediaType;
+  private final Date lastModified;
+  private final long size;
 
-  public RangeConsumesOutputStream(Path directAccessPath, String mediaType) {
-    this.directAccessPath = directAccessPath;
+  public RangeConsumesOutputStream(SeekableContentPayload payload, String filename, Date lastModified, long size,
+      String mediaType) {
+    this.payload = payload;
+    this.filename = filename;
+    this.lastModified = lastModified;
+    this.size = size;
     this.mediaType = mediaType;
   }
 
-  public RangeConsumesOutputStream(Path directAccessPath) {
-    this(directAccessPath, DEFAULT_MIME_TYPE);
+  public RangeConsumesOutputStream(SeekableContentPayload payload, Binary binary) {
+    this(payload, binary, DEFAULT_MIME_TYPE);
+  }
+
+  public RangeConsumesOutputStream(SeekableContentPayload payload, Binary binary, String mediaType) {
+    this.payload = payload;
+    this.filename = binary.getStoragePath().getName();
+    this.lastModified = new Date(); // TODO missing information about binary last modified date
+    this.size = binary.getSizeInBytes();
+    this.mediaType = mediaType;
   }
 
   @Override
   public void consumeOutputStream(OutputStream out) throws IOException {
-    try (InputStream in = Files.newInputStream(directAccessPath)) {
-      IOUtils.copyLarge(in, out);
-    }
+    payload.writeTo(out, 0, getSize());
   }
 
   @Override
   public void consumeOutputStream(OutputStream out, int from, int len) throws IOException {
-    try (InputStream in = Files.newInputStream(directAccessPath)) {
-      IOUtils.copyLarge(in, out, from, len);
-    }
+    payload.writeTo(out, from, len);
   }
 
   @Override
   public void consumeOutputStream(OutputStream out, long from, long end) {
     try {
-      File file = directAccessPath.toFile();
-      byte[] buffer = new byte[1024];
-      try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
-        long pos = from;
-        randomAccessFile.seek(pos);
-        while (pos < end) {
-          randomAccessFile.read(buffer);
-          out.write(buffer);
-          pos += buffer.length;
-        }
-        out.flush();
-      }
+      payload.writeTo(out, from, end - from);
     } catch (IOException e) {
-      // ignore
+      LOGGER.warn("Error writing to output stream", e);
     }
   }
 
   @Override
   public Date getLastModified() {
-    try {
-      return new Date(Files.getLastModifiedTime(directAccessPath).toMillis());
-    } catch (IOException e) {
-      return null;
-    }
+    return lastModified;
   }
 
   @Override
   public long getSize() {
-    try {
-      return Files.size(directAccessPath);
-    } catch (IOException e) {
-      return -1;
-    }
+    return size;
   }
 
   @Override
   public String getFileName() {
-    return directAccessPath.getFileName().toString();
+    return filename;
   }
 
   @Override

roda-core/roda-core/src/main/java/org/roda/core/storage/SeekableContentPayload.java

@@ -0,0 +1,14 @@
+package org.roda.core.storage;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+public interface SeekableContentPayload extends ContentPayload {
+    /**
+     * Writes a specific range of the content to the output stream.
+     * * @param out The output stream to write to
+     * @param offset The start byte position (inclusive)
+     * @param length The number of bytes to write
+     */
+    void writeTo(OutputStream out, long offset, long length) throws IOException;
+}

roda-core/roda-core/src/main/java/org/roda/core/storage/fs/FSPathContentPayload.java

@@ -9,19 +9,22 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 
-import org.roda.core.storage.ContentPayload;
+import org.apache.commons.io.IOUtils;
+import org.roda.core.storage.SeekableContentPayload;
 
 /**
  * Class that implements {@code ContentPayload} for File System
  *
  * @author Luis Faria <lfaria@keep.pt>
  */
-public class FSPathContentPayload implements ContentPayload {
+public class FSPathContentPayload implements SeekableContentPayload {
 
   private final Path path;
 
@@ -44,4 +47,41 @@ public URI getURI() throws IOException, UnsupportedOperationException {
     return path.toUri();
   }
 
+  @Override
+  public void writeTo(OutputStream out, long offset, long length) throws IOException {
+    // 1. Use NIO InputStream (Efficient: uses native pread/lseek)
+    try (InputStream is = Files.newInputStream(path)) {
+      
+      // 2. Seek to the start position (Instant operation on files)
+      long skipped = is.skip(offset);
+      if (skipped < offset) {
+        // File is smaller than the offset requested
+        return; 
+      }
+
+      // 3. Transfer only the requested amount
+      byte[] buffer = new byte[8192]; // Standard 8KB buffer
+      long remaining = length;
+      int bytesRead;
+
+      // Loop while we still need data AND we haven't hit EOF
+      while (remaining > 0) {
+        // Determine how much to read: either the full buffer or the remaining bytes
+        int bytesToRead = (int) Math.min(buffer.length, remaining);
+        
+        bytesRead = is.read(buffer, 0, bytesToRead);
+        
+        if (bytesRead == -1) {
+          break; // End of file reached prematurely
+        }
+
+        // Critical: Only write the bytes we actually read
+        out.write(buffer, 0, bytesRead);
+        
+        remaining -= bytesRead;
+      }
+      
+    }
+  }
+
 }

roda-ui/roda-wui/src/main/java/org/roda/wui/api/v2/services/DIPFileService.java

@@ -7,6 +7,10 @@
  */
 package org.roda.wui.api.v2.services;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
 import org.roda.core.data.exceptions.AuthorizationDeniedException;
 import org.roda.core.data.exceptions.GenericException;
 import org.roda.core.data.exceptions.NotFoundException;
@@ -17,29 +21,36 @@
 import org.roda.core.data.v2.ip.DIPFile;
 import org.roda.core.model.LiteRODAObjectFactory;
 import org.roda.core.model.ModelService;
+import org.roda.core.storage.Binary;
+import org.roda.core.storage.ContentPayload;
 import org.roda.core.storage.DirectResourceAccess;
 import org.roda.core.storage.RangeConsumesOutputStream;
+import org.roda.core.storage.SeekableContentPayload;
 import org.roda.wui.common.model.RequestContext;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
 /**
  *
  * @author Eduardo Teixeira <eteixeira@keep.pt>
  */
 @Service
 public class DIPFileService {
   public RangeConsumesOutputStream retrieveDIPFileRangeStream(RequestContext requestContext, DIPFile dipfile)
-    throws RequestNotValidException {
+      throws RequestNotValidException {
     ModelService model = requestContext.getModelService();
     if (!dipfile.isDirectory()) {
       final RangeConsumesOutputStream stream;
       try {
-        DirectResourceAccess directDIPFileAccess = model.getDirectAccess(dipfile);
-        stream = new RangeConsumesOutputStream(directDIPFileAccess.getPath());
+        Binary binary = model.getBinary(dipfile);
+        ContentPayload payload = binary.getContent();
+
+        if (payload instanceof SeekableContentPayload) {
+          SeekableContentPayload seekable = (SeekableContentPayload) payload;
+
+          stream = new RangeConsumesOutputStream(seekable, binary);
+        } else {
+          throw new RequestNotValidException("Range stream for file unsupported");
+        }
         return stream;
       } catch (RequestNotValidException | GenericException | AuthorizationDeniedException | NotFoundException e) {
         throw new RuntimeException(e);
@@ -50,7 +61,7 @@ public RangeConsumesOutputStream retrieveDIPFileRangeStream(RequestContext reque
   }
 
   public StreamResponse retrieveDIPFileStreamResponse(RequestContext requestContext, DIPFile dipFile)
-    throws GenericException, RequestNotValidException, NotFoundException, AuthorizationDeniedException {
+      throws GenericException, RequestNotValidException, NotFoundException, AuthorizationDeniedException {
     ModelService model = requestContext.getModelService();
 
     final ConsumesOutputStream stream;