Add IDataStreamReceiver.SaveToStreamAsync and lower in-memory threshold to 128MB

.gitignore

@@ -122,4 +122,6 @@ UpgradeLog*.XML
 
 tools/
 built-packages/**
-artifacts/
+artifacts/
+
+docs/superpowers/

source/Halibut.Tests/LocalDataStreamFixture.cs

@@ -1,12 +1,29 @@
 using System.IO;
 using System.Threading.Tasks;
 using FluentAssertions;
+using Halibut.Transport.Protocol;
 using NUnit.Framework;
 
 namespace Halibut.Tests
 {
     public class LocalDataStreamFixture : BaseTest
     {
+        [Test]
+        public void ShouldUseInMemoryReceiverForDataStreamsUnder128MB()
+        {
+            var dataStream = new DataStream(128 * 1024 * 1024 - 1, (stream, ct) => Task.CompletedTask);
+
+            dataStream.Receiver().Should().BeOfType<InMemoryDataStreamReceiver>();
+        }
+
+        [Test]
+        public void ShouldUseTemporaryFileReceiverForDataStreamsOf128MBOrOver()
+        {
+            var dataStream = new DataStream(128 * 1024 * 1024, (stream, ct) => Task.CompletedTask);
+
+            dataStream.Receiver().Should().BeOfType<TemporaryFileDataStreamReceiver>();
+        }
+
         [Test]
         public async Task ShouldUseInMemoryReceiverLocallyToRead()
         {

source/Halibut.Tests/Transport/Protocol/DataStreamReceiverSaveToStreamAsyncFixture.cs

@@ -0,0 +1,88 @@
+using System;
+using System.IO;
+using System.Text;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Halibut.Queue.Redis.MessageStorage;
+using Halibut.Tests.Support;
+using Halibut.Transport.Protocol;
+using NUnit.Framework;
+
+namespace Halibut.Tests.Transport.Protocol
+{
+    public class DataStreamReceiverSaveToStreamAsyncFixture : BaseTest
+    {
+        static byte[] SomeBytes() => Encoding.UTF8.GetBytes("Hello from SaveToStreamAsync!");
+
+        [Test]
+        public async Task InMemoryDataStreamReceiver_SaveToStreamAsync_WritesTheWritersData()
+        {
+            var data = SomeBytes();
+            var sut = new InMemoryDataStreamReceiver((stream, ct) => stream.WriteAsync(data, 0, data.Length, ct));
+
+            using var destination = new MemoryStream();
+            await sut.SaveToStreamAsync(destination, CancellationToken);
+
+            destination.ToArray().Should().BeEquivalentTo(data);
+        }
+
+        [Test]
+        public async Task TemporaryFileDataStreamReceiver_SaveToStreamAsync_WritesTheWritersData()
+        {
+            var data = SomeBytes();
+            var sut = new TemporaryFileDataStreamReceiver((stream, ct) => stream.WriteAsync(data, 0, data.Length, ct));
+
+            using var destination = new MemoryStream();
+            await sut.SaveToStreamAsync(destination, CancellationToken);
+
+            destination.ToArray().Should().BeEquivalentTo(data);
+        }
+
+        [Test]
+        public async Task TemporaryFileStream_SaveToStreamAsync_WritesTheFilesDataAndDeletesTheSourceFile()
+        {
+            var data = SomeBytes();
+            var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString());
+            File.WriteAllBytes(path, data);
+
+            var sut = new TemporaryFileStream(path, HalibutLog);
+
+            using var destination = new MemoryStream();
+            await sut.SaveToStreamAsync(destination, CancellationToken);
+
+            destination.ToArray().Should().BeEquivalentTo(data);
+            File.Exists(path).Should().BeFalse("the source temp file should be deleted once consumed");
+        }
+
+        [Test]
+        public async Task TemporaryFileStream_SaveToStreamAsync_CannotBeCalledTwice()
+        {
+            var data = SomeBytes();
+            var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString());
+            File.WriteAllBytes(path, data);
+
+            var sut = new TemporaryFileStream(path, HalibutLog);
+
+            using (var destination = new MemoryStream())
+            {
+                await sut.SaveToStreamAsync(destination, CancellationToken);
+            }
+
+            using var secondDestination = new MemoryStream();
+            await AssertException.Throws<InvalidOperationException>(async () => await sut.SaveToStreamAsync(secondDestination, CancellationToken));
+        }
+
+        [Test]
+        public async Task DataStreamRehydrationDataDataStreamReceiver_SaveToStreamAsync_WritesTheSuppliedData()
+        {
+            var data = SomeBytes();
+            var sourceStream = new MemoryStream(data);
+            var sut = new DataStreamRehydrationDataDataStreamReceiver(() => new DataStreamRehydrationData(sourceStream));
+
+            using var destination = new MemoryStream();
+            await sut.SaveToStreamAsync(destination, CancellationToken);
+
+            destination.ToArray().Should().BeEquivalentTo(data);
+        }
+    }
+}

source/Halibut/DataStream.cs

@@ -41,9 +41,9 @@ public IDataStreamReceiver Receiver()
                 return receiver;
             }
 
-            // Use a FileStream for packages over 2GB, or you risk running into OutOfMemory
+            // Use a FileStream for packages over 128MB, or you risk running into OutOfMemory
             // exceptions with MemoryStream.
-            var maxMemoryStreamLength = int.MaxValue;
+            const long maxMemoryStreamLength = 128 * 1024 * 1024;
             if (Length >= maxMemoryStreamLength)
             {
                 return new TemporaryFileDataStreamReceiver(writerAsync);

source/Halibut/IDataStreamReceiver.cs

@@ -10,5 +10,7 @@ public interface IDataStreamReceiver
         Task SaveToAsync(string filePath, CancellationToken cancellationToken);
 
         Task ReadAsync(Func<Stream, CancellationToken, Task> readerAsync, CancellationToken cancellationToken);
+
+        Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken);
     }
 }

source/Halibut/Queue/Redis/MessageStorage/IRehydrateDataStream.cs

@@ -65,19 +65,24 @@ public DataStreamRehydrationDataDataStreamReceiver(Func<DataStreamRehydrationDat
 
         public async Task SaveToAsync(string filePath, CancellationToken cancellationToken)
         {
-            await using var dataStreamRehydrationData = DataStreamRehydrationDataSupplier();
-
 #if !NETFRAMEWORK
             await
 #endif
                 using (var file = new FileStream(filePath, FileMode.Create))
             {
+                await SaveToStreamAsync(file, cancellationToken);
+            }
+        }
+
+        public async Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken)
+        {
+            await using var dataStreamRehydrationData = DataStreamRehydrationDataSupplier();
+
 #if NET8_0_OR_GREATER
-                await dataStreamRehydrationData.Data.CopyToAsync(file, cancellationToken);
+            await dataStreamRehydrationData.Data.CopyToAsync(destinationStream, cancellationToken);
 #else
-                await dataStreamRehydrationData.Data.CopyToAsync(file);
+            await dataStreamRehydrationData.Data.CopyToAsync(destinationStream);
 #endif
-            }
         }
 
         public async Task ReadAsync(Func<Stream, CancellationToken, Task> readerAsync, CancellationToken cancellationToken)

source/Halibut/Transport/Protocol/InMemoryDataStreamReceiver.cs

@@ -21,10 +21,16 @@ public async Task SaveToAsync(string filePath, CancellationToken cancellationTok
 #endif
                 using (var file = new FileStream(filePath, FileMode.Create))
             {
-                await writerAsync(file, cancellationToken);
+                await SaveToStreamAsync(file, cancellationToken);
             }
         }
 
+        public async Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken)
+        {
+            await writerAsync(destinationStream, cancellationToken);
+
+        }
+
         public async Task ReadAsync(Func<Stream, CancellationToken, Task> readerAsync, CancellationToken cancellationToken)
         {
             using (var stream = new MemoryStream())

source/Halibut/Transport/Protocol/TemporaryFileDataStreamReceiver.cs

@@ -21,10 +21,15 @@ public async Task SaveToAsync(string filePath, CancellationToken cancellationTok
 #endif
             using (var file = new FileStream(filePath, FileMode.Create))
             {
-                await writerAsync(file, cancellationToken);
+                await SaveToStreamAsync(file, cancellationToken);
             }
         }
-
+
+        public async Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken)
+        {
+            await writerAsync(destinationStream, cancellationToken);
+        }
+
         public async Task ReadAsync(Func<Stream, CancellationToken, Task> readerAsync, CancellationToken cancellationToken)
         {
             var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString());

source/Halibut/Transport/Protocol/TemporaryFileStream.cs

@@ -62,6 +62,23 @@ void SetFilePermissionsToInheritFromParent(string filePath)
             }
         }
 
+        public async Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken)
+        {
+            if (moved) throw new InvalidOperationException("This stream has already been received once, and it cannot be read again.");
+
+            using (var file = new FileStream(path, FileMode.Open, FileAccess.Read))
+            {
+#if NET8_0_OR_GREATER
+                await file.CopyToAsync(destinationStream, cancellationToken);
+#else
+                await file.CopyToAsync(destinationStream);
+#endif
+            }
+            await AttemptToDeleteAsync(path);
+            moved = true;
+            GC.SuppressFinalize(this);
+        }
+
         public async Task ReadAsync(Func<Stream, CancellationToken, Task> readerAsync, CancellationToken cancellationToken)
         {
             if (moved) throw new InvalidOperationException("This stream has already been received once, and it cannot be read again.");