fix: Don't dispose source stream by default in ToMemoryStreamAsync (#1724)

* fix: Don't dispose source stream by default in ToMemoryStreamAsync

Fixes #1534

The ToMemoryStreamAsync extension method was unexpectedly disposing the input
stream after copying, which is surprising behavior for an extension method.
Additionally, behavior was inconsistent - MemoryStream inputs were returned
as-is (not disposed), while other streams were disposed.

Changes:
- Add disposeSource parameter (defaults to false)
- Reset output MemoryStream position to 0 for immediate reading
- Also reset MemoryStream input position to 0 for consistency
- Updated tests to pass disposeSource: true to prevent resource leaks

Callers who want the old disposal behavior can pass disposeSource: true.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Handle disposeSource consistently for MemoryStream inputs

When disposeSource=true and input is a MemoryStream, now copies to a new
MemoryStream before disposing the original. This ensures consistent behavior
regardless of input stream type.

Before: MemoryStream inputs were returned directly, ignoring disposeSource
After: MemoryStream inputs are copied when disposeSource=true, then disposed

* fix: Use await using for MemoryStream in tests

Use await using declaration for MemoryStream results in tests for
consistency and to ensure proper disposal patterns.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* docs: Add detailed remarks for ToMemoryStreamAsync breaking change

Document:
- Breaking change from previous behavior (now defaults to no disposal)
- MemoryStream optimization behavior
- Caller responsibility for disposing returned stream

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: thomhurst

src/ModularPipelines/Extensions/StreamExtensions.cs

@@ -8,16 +8,44 @@ public static class StreamExtensions
     /// <summary>
     /// Turns a generic <see cref="Stream"/> into a <see cref="MemoryStream"/>.
     /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Breaking change:</b> This method previously always disposed the source stream.
+    /// It now defaults to <c>disposeSource: false</c> to follow the principle of least surprise.
+    /// Existing callers that relied on automatic disposal should pass <c>disposeSource: true</c>.
+    /// </para>
+    /// <para>
+    /// When the input is already a <see cref="MemoryStream"/> and <paramref name="disposeSource"/> is false,
+    /// the same instance is returned (with position reset) as an optimization.
+    /// When <paramref name="disposeSource"/> is true, a copy is made before disposing the source.
+    /// </para>
+    /// <para>
+    /// Callers are responsible for disposing the returned <see cref="MemoryStream"/>.
+    /// </para>
+    /// </remarks>
     /// <param name="stream">Any stream.</param>
-    /// <returns>A MemoryStream containing the Stream's data.</returns>
-    public static async Task<MemoryStream> ToMemoryStreamAsync(this Stream stream)
+    /// <param name="disposeSource">
+    /// When true, disposes the source stream after copying.
+    /// Defaults to false - callers are responsible for disposing the source stream.
+    /// </param>
+    /// <returns>A MemoryStream containing the Stream's data, with Position set to 0 for reading.</returns>
+    public static async Task<MemoryStream> ToMemoryStreamAsync(this Stream stream, bool disposeSource = false)
     {
-        if (stream is MemoryStream memoryStream)
+        // If input is already a MemoryStream and we don't need to dispose it,
+        // return it directly (optimization to avoid unnecessary copy)
+        if (stream is MemoryStream sourceMs && !disposeSource)
         {
-            return memoryStream;
+            if (sourceMs.CanSeek)
+            {
+                sourceMs.Position = 0;
+            }
+
+            return sourceMs;
         }
 
-        memoryStream = new MemoryStream();
+        // Copy to new MemoryStream (handles both non-MemoryStream inputs
+        // and MemoryStream inputs when disposeSource is true)
+        var memoryStream = new MemoryStream();
 
         if (stream.CanSeek)
         {
@@ -26,8 +54,12 @@ public static async Task<MemoryStream> ToMemoryStreamAsync(this Stream stream)
 
         await stream.CopyToAsync(memoryStream).ConfigureAwait(false);
 
-        await stream.DisposeAsync().ConfigureAwait(false);
+        if (disposeSource)
+        {
+            await stream.DisposeAsync().ConfigureAwait(false);
+        }
 
+        memoryStream.Position = 0;
         return memoryStream;
     }
 }

test/ModularPipelines.UnitTests/FileTests.cs

@@ -107,7 +107,7 @@ public async Task ReadEmptyFile()
         var plainText = await file.ReadAsync();
         var lines = await ToListAsync(file.ReadLinesAsync());
         var bytes = await file.ReadBytesAsync();
-        var stream = await file.GetStream().ToMemoryStreamAsync();
+        await using var stream = await file.GetStream().ToMemoryStreamAsync(disposeSource: true);
 
         using (Assert.Multiple())
         {
@@ -128,7 +128,7 @@ public async Task ReadWriteFile()
         var plainText = await file.ReadAsync();
         var lines = await ToListAsync(file.ReadLinesAsync());
         var bytes = await file.ReadBytesAsync();
-        var stream = await file.GetStream().ToMemoryStreamAsync();
+        await using var stream = await file.GetStream().ToMemoryStreamAsync(disposeSource: true);
 
         using (Assert.Multiple())
         {