fix: 110: ReadableStreamingData limit is not set if it's wrapped over…

… a file stream

Provided a new constructor to create a ReadableStreamingData from a file path
Provided a new constructor to create a ReadableStreamingData from a byte array

Signed-off-by: Artem Ananev <artem.ananev@swirldslabs.com>

pbj-core/pbj-runtime/src/main/java/com/hedera/pbj/runtime/io/stream/ReadableStreamingData.java

@@ -4,10 +4,14 @@
 import com.hedera.pbj.runtime.io.ReadableSequentialData;
 import com.hedera.pbj.runtime.io.buffer.BufferedData;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
 
 import static java.util.Objects.requireNonNull;
 
@@ -27,14 +31,36 @@ public class ReadableStreamingData implements ReadableSequentialData, AutoClosea
     private boolean eof = false;
 
     /**
-     * Creates a {@code FilterInputStream} that implements {@code DataInput} API.
+     * Creates a new streaming data object on top of a given input stream.
      *
      * @param in the underlying input stream, can not be null
      */
     public ReadableStreamingData(@NonNull final InputStream in) {
         this.in = requireNonNull(in);
     }
 
+    /**
+     * Opens a new input stream to read a given file and creates a new streaming data object on top
+     * of this stream.
+     *
+     * @param file the file, can not be null
+     * @throws IOException if an I/O error occurs
+     */
+    public ReadableStreamingData(@NonNull final Path file) throws IOException {
+        this.in = Files.newInputStream(file, StandardOpenOption.READ);
+        this.limit = Files.size(file);
+    }
+
+    /**
+     * Creates a new streaming data object on top of a given byte array.
+     *
+     * @param bytes the byte array, can not be null
+     */
+    public ReadableStreamingData(@NonNull final byte[] bytes) {
+        this.in = new ByteArrayInputStream(bytes);
+        this.limit = bytes.length;
+    }
+
     // ================================================================================================================
     // AutoCloseable Methods
 

pbj-core/pbj-runtime/src/test/java/com/hedera/pbj/runtime/io/stream/ReadableStreamingDataTest.java

@@ -3,6 +3,9 @@
 import com.hedera.pbj.runtime.io.DataAccessException;
 import com.hedera.pbj.runtime.io.ReadableSequentialTestBase;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import java.io.BufferedInputStream;
@@ -21,7 +24,7 @@ final class ReadableStreamingDataTest extends ReadableSequentialTestBase {
     @NonNull
     @Override
     protected ReadableStreamingData emptySequence() {
-        final var stream = new ReadableStreamingData(new ByteArrayInputStream(new byte[0]));
+        final var stream = new ReadableStreamingData(new byte[0]);
         stream.limit(0);
         return stream;
     }
@@ -39,7 +42,7 @@ protected ReadableStreamingData fullyUsedSequence() {
     @Override
     @NonNull
     protected ReadableStreamingData sequence(@NonNull byte [] arr) {
-        final var stream = new ReadableStreamingData(new ByteArrayInputStream(arr));
+        final var stream = new ReadableStreamingData(arr);
         stream.limit(arr.length);
         return stream;
     }
@@ -94,7 +97,7 @@ public synchronized long skip(long n) throws IOException {
 
     @Test
     @DisplayName("Bad InputStream will fail on read")
-    void inputStreamFailsDuringRead() throws IOException {
+    void inputStreamFailsDuringRead() {
         final var throwNow = new AtomicBoolean(false);
         final var byteStream = new ByteArrayInputStream(new byte[] { 1, 2, 3, 4, 5, 6, 7 });
         final var inputStream = new BufferedInputStream(byteStream) {
@@ -180,4 +183,32 @@ void limitNotChanged() {
         }
     }
 
+    @Test
+    void zeroLimitReadEmptyFile() throws IOException {
+        final Path file = Files.createTempFile(getClass().getSimpleName(), "zeroLimitReadEmptyFile");
+        try (final var stream = new ReadableStreamingData(file)) {
+            assertThat(stream.limit()).isEqualTo(0);
+        }
+    }
+
+    @Test
+    void nonZeroLimitReadFile() throws IOException {
+        final Path file = Files.createTempFile(getClass().getSimpleName(), "nonZeroLimitReadFile");
+        final byte[] bytes = new byte[] {0, 1, 2, 3, 4};
+        Files.write(file, bytes, StandardOpenOption.WRITE);
+        try (final var stream = new ReadableStreamingData(file)) {
+            assertThat(stream.limit()).isEqualTo(bytes.length);
+        }
+    }
+
+    @Test
+    void dataOnTopOfByteArrayLimits() {
+        final byte[] bytes = new byte[] {0, 1, 2, 3, 4, 5, 6, 7};
+        try (final var stream = new ReadableStreamingData(bytes)) {
+            assertThat(stream.limit()).isEqualTo(bytes.length);
+        }
+        try (final var stream = new ReadableStreamingData(new ByteArrayInputStream(bytes))) {
+            assertThat(stream.limit()).isEqualTo(Long.MAX_VALUE);
+        }
+    }
 }