Add isAvailable to check if has not been consumed

Author: mtdowling

io/src/main/java/software/amazon/smithy/java/io/datastream/ByteBufferDataStream.java

@@ -31,6 +31,11 @@ public boolean isReplayable() {
         return true;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return true;
+    }
+
     @Override
     public ByteBuffer asByteBuffer() {
         return buffer.duplicate();

io/src/main/java/software/amazon/smithy/java/io/datastream/DataStream.java

@@ -57,6 +57,21 @@ default boolean hasKnownLength() {
      */
     boolean isReplayable();
 
+    /**
+     * Check if the DataStream is available for consumption.
+     *
+     * <p>A stream is available if it either hasn't been consumed yet, or if it is replayable. This is useful for
+     * interceptors that need to make decisions based on whether the stream data can be accessed without causing
+     * an error. In other worse, if this method returns {@code true}, then calling {@link #asInputStream()} will
+     * succeed.
+     *
+     * <p>Note: This method returning {@code true} does not guarantee that reading will succeed (e.g., I/O errors
+     * can still occur), only that the stream has not been previously consumed in a non-replayable manner.
+     *
+     * @return true if the stream can be consumed.
+     */
+    boolean isAvailable();
+
     /**
      * Convert the stream into a blocking {@link InputStream}.
      *

io/src/main/java/software/amazon/smithy/java/io/datastream/EmptyDataStream.java

@@ -31,6 +31,11 @@ public boolean isReplayable() {
         return true;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return true;
+    }
+
     @Override
     public long contentLength() {
         return 0;

io/src/main/java/software/amazon/smithy/java/io/datastream/FileDataStream.java

@@ -46,6 +46,11 @@ public boolean isReplayable() {
         return true;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return true;
+    }
+
     @Override
     public InputStream asInputStream() {
         try {

io/src/main/java/software/amazon/smithy/java/io/datastream/InputStreamDataStream.java

@@ -34,6 +34,11 @@ public boolean isReplayable() {
         return false;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return !consumed;
+    }
+
     @Override
     public long contentLength() {
         return contentLength;

io/src/main/java/software/amazon/smithy/java/io/datastream/PublisherDataStream.java

@@ -40,6 +40,11 @@ public boolean isReplayable() {
         return isReplayable;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return isReplayable || !consumed;
+    }
+
     // Override to skip needing an intermediate InputStream for this.
     @Override
     public ByteBuffer asByteBuffer() {

io/src/main/java/software/amazon/smithy/java/io/datastream/WrappedDataStream.java

@@ -48,6 +48,11 @@ public boolean isReplayable() {
         return isReplayable;
     }
 
+    @Override
+    public boolean isAvailable() {
+        return delegate.isAvailable();
+    }
+
     @Override
     public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
         delegate.subscribe(subscriber);

io/src/test/java/software/amazon/smithy/java/io/datastream/ByteBufferDataStreamTest.java

@@ -22,4 +22,13 @@ public void returnsByteBuffer() {
         assertThat(ds.asByteBuffer(), equalTo(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8))));
         assertThat(ds.isReplayable(), is(true));
     }
+
+    @Test
+    public void isAlwaysAvailable() {
+        var ds = DataStream.ofBytes("foo".getBytes(StandardCharsets.UTF_8));
+
+        assertThat(ds.isAvailable(), is(true));
+        ds.asByteBuffer();
+        assertThat(ds.isAvailable(), is(true));
+    }
 }

io/src/test/java/software/amazon/smithy/java/io/datastream/FileDataStreamTest.java

@@ -47,4 +47,13 @@ public void readsDataToByteBuffer() throws Exception {
 
         assertThat(ds.asByteBuffer(), equalTo(ByteBuffer.wrap("Hello!".getBytes(StandardCharsets.UTF_8))));
     }
+
+    @Test
+    public void isAlwaysAvailable() throws Exception {
+        var ds = DataStream.ofFile(Paths.get(getClass().getResource("test.txt").toURI()));
+
+        assertThat(ds.isAvailable(), is(true));
+        ds.asByteBuffer();
+        assertThat(ds.isAvailable(), is(true));
+    }
 }

io/src/test/java/software/amazon/smithy/java/io/datastream/InputStreamDataStreamTest.java

@@ -70,4 +70,21 @@ public void cannotReadDataStreamTwice() throws Exception {
         Assertions.assertDoesNotThrow(ds::asInputStream);
         Assertions.assertThrows(IllegalStateException.class, ds::asInputStream);
     }
+
+    @Test
+    public void isAvailableBeforeConsumption() throws Exception {
+        var ds = DataStream.ofInputStream(
+                Files.newInputStream(Paths.get(getClass().getResource("test.txt").toURI())));
+
+        assertThat(ds.isAvailable(), is(true));
+    }
+
+    @Test
+    public void isNotAvailableAfterConsumption() throws Exception {
+        var ds = DataStream.ofInputStream(
+                Files.newInputStream(Paths.get(getClass().getResource("test.txt").toURI())));
+
+        ds.asInputStream();
+        assertThat(ds.isAvailable(), is(false));
+    }
 }