diff --git a/.gitignore b/.gitignore index 30d1d0cf..31ae1fac 100644 --- a/.gitignore +++ b/.gitignore @@ -122,4 +122,6 @@ UpgradeLog*.XML tools/ built-packages/** -artifacts/ \ No newline at end of file +artifacts/ + +docs/superpowers/ \ No newline at end of file diff --git a/source/Halibut.Tests/LocalDataStreamFixture.cs b/source/Halibut.Tests/LocalDataStreamFixture.cs index e03c7c7c..b1cde9e8 100644 --- a/source/Halibut.Tests/LocalDataStreamFixture.cs +++ b/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(); + } + + [Test] + public void ShouldUseTemporaryFileReceiverForDataStreamsOf128MBOrOver() + { + var dataStream = new DataStream(128 * 1024 * 1024, (stream, ct) => Task.CompletedTask); + + dataStream.Receiver().Should().BeOfType(); + } + [Test] public async Task ShouldUseInMemoryReceiverLocallyToRead() { diff --git a/source/Halibut.Tests/Transport/Protocol/DataStreamReceiverSaveToStreamAsyncFixture.cs b/source/Halibut.Tests/Transport/Protocol/DataStreamReceiverSaveToStreamAsyncFixture.cs new file mode 100644 index 00000000..738a7b80 --- /dev/null +++ b/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(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); + } + } +} diff --git a/source/Halibut/DataStream.cs b/source/Halibut/DataStream.cs index 8b8dea19..6b616727 100644 --- a/source/Halibut/DataStream.cs +++ b/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); diff --git a/source/Halibut/IDataStreamReceiver.cs b/source/Halibut/IDataStreamReceiver.cs index 5a66c0ff..60316054 100644 --- a/source/Halibut/IDataStreamReceiver.cs +++ b/source/Halibut/IDataStreamReceiver.cs @@ -10,5 +10,7 @@ public interface IDataStreamReceiver Task SaveToAsync(string filePath, CancellationToken cancellationToken); Task ReadAsync(Func readerAsync, CancellationToken cancellationToken); + + Task SaveToStreamAsync(Stream destinationStream, CancellationToken cancellationToken); } } \ No newline at end of file diff --git a/source/Halibut/Queue/Redis/MessageStorage/IRehydrateDataStream.cs b/source/Halibut/Queue/Redis/MessageStorage/IRehydrateDataStream.cs index 3f844041..8591d9ab 100644 --- a/source/Halibut/Queue/Redis/MessageStorage/IRehydrateDataStream.cs +++ b/source/Halibut/Queue/Redis/MessageStorage/IRehydrateDataStream.cs @@ -65,19 +65,24 @@ public DataStreamRehydrationDataDataStreamReceiver(Func readerAsync, CancellationToken cancellationToken) diff --git a/source/Halibut/Transport/Protocol/InMemoryDataStreamReceiver.cs b/source/Halibut/Transport/Protocol/InMemoryDataStreamReceiver.cs index 996ee7d9..ee62a216 100644 --- a/source/Halibut/Transport/Protocol/InMemoryDataStreamReceiver.cs +++ b/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 readerAsync, CancellationToken cancellationToken) { using (var stream = new MemoryStream()) diff --git a/source/Halibut/Transport/Protocol/TemporaryFileDataStreamReceiver.cs b/source/Halibut/Transport/Protocol/TemporaryFileDataStreamReceiver.cs index ccd07e48..b09c2cc0 100644 --- a/source/Halibut/Transport/Protocol/TemporaryFileDataStreamReceiver.cs +++ b/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 readerAsync, CancellationToken cancellationToken) { var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString()); diff --git a/source/Halibut/Transport/Protocol/TemporaryFileStream.cs b/source/Halibut/Transport/Protocol/TemporaryFileStream.cs index ee6bf0c2..a0bb9942 100644 --- a/source/Halibut/Transport/Protocol/TemporaryFileStream.cs +++ b/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 readerAsync, CancellationToken cancellationToken) { if (moved) throw new InvalidOperationException("This stream has already been received once, and it cannot be read again.");