Add close to DataStream to allow closing resources

Author: mtdowling

io/src/main/java/software/amazon/smithy/java/io/ByteBufferOutputStream.java

@@ -68,6 +68,14 @@ public int size() {
         return count;
     }
 
+    /**
+     * Resets the stream to empty, allowing buffer reuse.
+     * The internal buffer is retained, avoiding reallocation.
+     */
+    public void reset() {
+        count = 0;
+    }
+
     private void ensureCapacity(int minCapacity) {
         // overflow-conscious code
         int oldCapacity = buf.length;

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

@@ -19,7 +19,7 @@
 /**
  * Abstraction for reading streams of data.
  */
-public interface DataStream extends Flow.Publisher<ByteBuffer> {
+public interface DataStream extends Flow.Publisher<ByteBuffer>, AutoCloseable {
     /**
      * Length of the data stream, if known.
      *
@@ -105,6 +105,19 @@ default void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
         HttpRequest.BodyPublishers.ofInputStream(this::asInputStream).subscribe(subscriber);
     }
 
+    /**
+     * Closes any underlying resources associated with this data stream.
+     *
+     * <p>The default implementation does nothing. Implementations that hold closeable resources (e.g., input streams)
+     * should override this method to release them.
+     *
+     * <p>It is safe to call this method multiple times.
+     */
+    @Override
+    default void close() {
+        // Default no-op. Implementations holding closeable resources should override.
+    }
+
     /**
      * Create an empty DataStream.
      *

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

@@ -5,14 +5,17 @@
 
 package software.amazon.smithy.java.io.datastream;
 
+import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 
 final class InputStreamDataStream implements DataStream {
 
     private final InputStream inputStream;
     private final String contentType;
     private final long contentLength;
     private boolean consumed;
+    private boolean closed;
 
     InputStreamDataStream(InputStream inputStream, String contentType, long contentLength) {
         this.inputStream = inputStream;
@@ -48,4 +51,16 @@ public long contentLength() {
     public String contentType() {
         return contentType;
     }
+
+    @Override
+    public void close() {
+        if (!closed) {
+            closed = true;
+            try {
+                inputStream.close();
+            } catch (IOException e) {
+                throw new UncheckedIOException("Failed to close data stream", e);
+            }
+        }
+    }
 }

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

@@ -57,4 +57,9 @@ public boolean isAvailable() {
     public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
         delegate.subscribe(subscriber);
     }
+
+    @Override
+    public void close() {
+        delegate.close();
+    }
 }