diff --git a/src/libraries/Common/src/System/Net/StreamBuffer.cs b/src/libraries/Common/src/System/Net/StreamBuffer.cs
index e8ea8f961d85ad..aa7c74622b07d2 100644
--- a/src/libraries/Common/src/System/Net/StreamBuffer.cs
+++ b/src/libraries/Common/src/System/Net/StreamBuffer.cs
@@ -343,7 +343,6 @@ public void Reset()
{
if (_hasWaiter != 0)
{
- Debug.Fail("Concurrent use is not supported");
throw new InvalidOperationException("Concurrent use is not supported");
}
diff --git a/src/libraries/Common/tests/Common.Tests.csproj b/src/libraries/Common/tests/Common.Tests.csproj
index b7c9f6b3ebabc4..12e815eb4a5535 100644
--- a/src/libraries/Common/tests/Common.Tests.csproj
+++ b/src/libraries/Common/tests/Common.Tests.csproj
@@ -102,6 +102,12 @@
Link="Common\System\Threading\Tasks\TaskToApm.cs" />
+
+
+
+
+
+
UncompressedTestFiles()
{
@@ -36,8 +38,22 @@ public abstract class CompressionStreamTestBase : CompressionTestBase
public abstract Stream CreateStream(Stream stream, CompressionLevel level);
public abstract Stream CreateStream(Stream stream, CompressionLevel level, bool leaveOpen);
public abstract Stream BaseStream(Stream stream);
- public virtual bool FlushCompletes { get => true; }
- public virtual bool FlushNoOps { get => false; }
public virtual int BufferSize { get => 8192; }
+
+ protected override Task CreateConnectedStreamsAsync()
+ {
+ (Stream stream1, Stream stream2) = ConnectedStreams.CreateBidirectional(4 * 1024, 16 * 1024);
+ return Task.FromResult((CreateStream(stream1, CompressionMode.Compress), CreateStream(stream2, CompressionMode.Decompress)));
+ }
+
+ protected override Task CreateWrappedConnectedStreamsAsync(StreamPair wrapped, bool leaveOpen) =>
+ Task.FromResult((CreateStream(wrapped.Stream1, CompressionMode.Compress, leaveOpen), CreateStream(wrapped.Stream2, CompressionMode.Decompress, leaveOpen)));
+
+ protected override int BufferedSize => 16 * 1024 + BufferSize;
+ protected override bool UsableAfterCanceledReads => false;
+ protected override Type UnsupportedReadWriteExceptionType => typeof(InvalidOperationException);
+ protected override bool WrappedUsableAfterClose => false;
+ protected override bool FlushRequiredToWriteData => true;
+ protected override bool FlushGuaranteesAllDataWritten => false;
}
}
diff --git a/src/libraries/Common/tests/System/IO/Compression/CompressionStreamUnitTestBase.cs b/src/libraries/Common/tests/System/IO/Compression/CompressionStreamUnitTestBase.cs
index 96db35c4be9a37..7e81c077709312 100644
--- a/src/libraries/Common/tests/System/IO/Compression/CompressionStreamUnitTestBase.cs
+++ b/src/libraries/Common/tests/System/IO/Compression/CompressionStreamUnitTestBase.cs
@@ -2,7 +2,6 @@
// The .NET Foundation licenses this file to you under the MIT license.
using System.Collections.Generic;
-using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
@@ -17,8 +16,6 @@ public abstract class CompressionStreamUnitTestBase : CompressionStreamTestBase
[ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
public virtual void FlushAsync_DuringWriteAsync()
{
- if (FlushNoOps)
- return;
byte[] buffer = new byte[100000];
Random rand = new Random();
rand.NextBytes(buffer);
@@ -52,8 +49,6 @@ public virtual void FlushAsync_DuringWriteAsync()
[ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
public async Task FlushAsync_DuringReadAsync()
{
- if (FlushNoOps)
- return;
byte[] buffer = new byte[32];
string testFilePath = CompressedTestFile(UncompressedTestFile());
using (var readStream = await ManualSyncMemoryStream.GetStreamFromFileAsync(testFilePath, false))
@@ -81,8 +76,6 @@ public async Task FlushAsync_DuringReadAsync()
[ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
public async Task FlushAsync_DuringFlushAsync()
{
- if (FlushNoOps)
- return;
byte[] buffer = null;
string testFilePath = CompressedTestFile(UncompressedTestFile());
using (var origStream = await LocalMemoryStream.readAppFileAsync(testFilePath))
@@ -121,64 +114,6 @@ public async Task FlushAsync_DuringFlushAsync()
}
}
- [ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
- public virtual void WriteAsync_DuringWriteAsync()
- {
- byte[] buffer = new byte[100000];
- Random rand = new Random();
- rand.NextBytes(buffer);
-
- using (var writeStream = new ManualSyncMemoryStream(false))
- using (var compressor = CreateStream(writeStream, CompressionMode.Compress))
- {
- Task task = null;
- try
- {
- // Write needs to be big enough to trigger a write to the underlying base stream so the WriteAsync call doesn't immediately complete.
- task = compressor.WriteAsync(buffer, 0, buffer.Length);
- while (task.IsCompleted)
- {
- rand.NextBytes(buffer);
- task = compressor.WriteAsync(buffer, 0, buffer.Length);
- }
- Assert.Throws(() => { compressor.WriteAsync(buffer, 32, 32); }); // "overlapping write"
- }
- finally
- {
- // Unblock Async operations
- writeStream.manualResetEvent.Set();
- // The original WriteAsync should be able to complete
- Assert.True(task.Wait(TaskTimeout), "Original WriteAsync Task did not complete in time");
- Assert.True(writeStream.WriteHit, "BaseStream Write function was not called");
- }
- }
- }
-
- [ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
- public async Task ReadAsync_DuringReadAsync()
- {
- byte[] buffer = new byte[32];
- string testFilePath = CompressedTestFile(UncompressedTestFile());
- using (var readStream = await ManualSyncMemoryStream.GetStreamFromFileAsync(testFilePath, false))
- using (var decompressor = CreateStream(readStream, CompressionMode.Decompress, true))
- {
- Task task = null;
- try
- {
- task = decompressor.ReadAsync(buffer, 0, 32);
- Assert.Throws(() => { decompressor.ReadAsync(buffer, 0, 32); }); // "overlapping read"
- }
- finally
- {
- // Unblock Async operations
- readStream.manualResetEvent.Set();
- // The original ReadAsync should be able to complete
- Assert.True(task.Wait(TaskTimeout), "The original ReadAsync should be able to complete");
- Assert.True(readStream.ReadHit, "BaseStream ReadAsync should have been called");
- }
- }
- }
-
[ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
public virtual async Task Dispose_WithUnfinishedReadAsync()
{
@@ -287,20 +222,6 @@ public async Task Read_BaseStreamSlowly()
}
}
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void CanReadCanWrite(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- Assert.True(mode == CompressionMode.Compress ? compressor.CanWrite : compressor.CanRead);
-
- compressor.Dispose();
- Assert.False(compressor.CanRead);
- Assert.False(compressor.CanWrite);
- }
-
[Theory]
[InlineData(CompressionMode.Compress)]
[InlineData(CompressionMode.Decompress)]
@@ -332,7 +253,7 @@ public void Ctor_DisposedBaseStream(CompressionMode mode)
[Theory]
[InlineData(CompressionMode.Compress)]
[InlineData(CompressionMode.Decompress)]
- public void Ctor_InvalidStreamCanReadCanWrite(CompressionMode mode)
+ public void Ctor_InvalidStream_Throws(CompressionMode mode)
{
LocalMemoryStream ms = new LocalMemoryStream();
ms.SetCanRead(mode == CompressionMode.Compress);
@@ -341,27 +262,26 @@ public void Ctor_InvalidStreamCanReadCanWrite(CompressionMode mode)
AssertExtensions.Throws("stream", () => CreateStream(ms, mode));
}
- public IEnumerable> CtorFunctions()
- {
- CompressionLevel[] legalValues = new CompressionLevel[] { CompressionLevel.Optimal, CompressionLevel.Fastest, CompressionLevel.NoCompression };
- yield return new Func((stream) => CreateStream(stream, CompressionMode.Compress));
- foreach (CompressionLevel level in legalValues)
- {
- yield return new Func((stream) => CreateStream(stream, level));
- bool[] boolValues = new bool[] { true, false };
+ [Fact]
+ public void TestCompressCtor()
+ {
+ IEnumerable> CtorFunctions()
+ {
+ yield return new Func((stream) => CreateStream(stream, CompressionMode.Compress));
- foreach (bool remainsOpen in boolValues)
+ foreach (CompressionLevel level in new[] { CompressionLevel.Optimal, CompressionLevel.Fastest, CompressionLevel.NoCompression })
{
- yield return new Func((stream) => CreateStream(stream, level, remainsOpen));
+ yield return new Func((stream) => CreateStream(stream, level));
+
+ foreach (bool remainsOpen in new[] { true, false })
+ {
+ yield return new Func((stream) => CreateStream(stream, level, remainsOpen));
+ }
}
}
- }
- [Fact]
- public void TestCompressCtor()
- {
Assert.All(CtorFunctions(), (create) =>
{
//Create the Stream
@@ -400,41 +320,6 @@ public void TestCompressCtor()
});
}
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void TestLeaveOpen(CompressionMode mode)
- {
- //Create the Stream
- var baseStream = new MemoryStream();
- Stream compressor = CreateStream(baseStream, mode, leaveOpen: false);
- compressor.Dispose();
-
- //Check that Close has really closed the underlying stream
- Assert.Throws(() => baseStream.Write(new byte[] { }, 0, 0));
- }
-
- [Fact]
- public void TestLeaveOpenAfterValidCompress()
- {
- //Create the Stream
- int _bufferSize = 1024;
- var bytes = new byte[_bufferSize];
- var baseStream = new MemoryStream(bytes, writable: true);
- Stream compressor = CreateStream(baseStream, CompressionMode.Compress, leaveOpen: false);
-
- //Write some data and Close the stream
- string strData = "Test Data";
- var encoding = Encoding.UTF8;
- byte[] data = encoding.GetBytes(strData);
- compressor.Write(data, 0, data.Length);
- compressor.Flush();
- compressor.Dispose();
-
- //Check that Close has really closed the underlying stream
- Assert.Throws(() => baseStream.Write(bytes, 0, bytes.Length));
- }
-
[Fact]
public async Task TestLeaveOpenAfterValidDecompress()
{
@@ -470,619 +355,6 @@ public void Ctor_ArgumentValidation()
AssertExtensions.Throws("stream", () => CreateStream(new MemoryStream(new byte[1], writable: false), CompressionLevel.Optimal));
}
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public async Task Flush(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Flush();
- await compressor.FlushAsync();
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void Flush_Double(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Flush();
- compressor.Flush();
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void Dispose_Double(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Dispose();
- compressor.Dispose();
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void Flush_FollowedByDispose(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Flush();
- compressor.Dispose();
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void Dispose_FollowedBySyncOperations(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Dispose();
-
- if (mode == CompressionMode.Compress)
- Assert.Throws(() => compressor.Write(new byte[1], 0, 1));
- else
- Assert.Throws(() => compressor.Read(new byte[1], 0, 1));
- Assert.Throws(() => compressor.Flush());
- Assert.Throws(() => compressor.CopyTo(new MemoryStream()));
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public virtual async Task Dispose_FollowedByAsyncOperations(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var compressor = CreateStream(ms, mode);
- compressor.Dispose();
-
- if (mode == CompressionMode.Compress)
- await Assert.ThrowsAsync(async () => await compressor.WriteAsync(new byte[1], 0, 1));
- else
- await Assert.ThrowsAsync(async () => await compressor.ReadAsync(new byte[1], 0, 1));
- await Assert.ThrowsAsync(async () => await compressor.FlushAsync());
- await Assert.ThrowsAsync(async () => await compressor.CopyToAsync(new MemoryStream()));
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void TestSeekMethods(CompressionMode mode)
- {
- var ms = new MemoryStream();
- var decompressor = CreateStream(ms, mode);
- Assert.False(decompressor.CanSeek, "CanSeek should be false");
- Assert.Throws(() => decompressor.Length);
- Assert.Throws(() => decompressor.SetLength(1));
- Assert.Throws(() => decompressor.Position);
- Assert.Throws(() => decompressor.Position = 100L);
- Assert.Throws(() => decompressor.Seek(100L, SeekOrigin.Begin));
- }
-
- [Theory]
- [InlineData(false)]
- [InlineData(true)]
- public virtual void Write_ArgumentValidation(bool useAsync)
- {
- using (var decompressor = CreateStream(new MemoryStream(), CompressionMode.Compress))
- {
- Assert.Throws(() => { if (useAsync) { decompressor.WriteAsync(null, 0, 0).Wait(); } else { decompressor.Write(null, 0, 0); } });
- Assert.Throws(() => { if (useAsync) { decompressor.WriteAsync(new byte[1], -1, 0).Wait(); } else { decompressor.Write(new byte[1], -1, 0); } });
- Assert.Throws(() => { if (useAsync) { decompressor.WriteAsync(new byte[1], 0, -1).Wait(); } else { decompressor.Write(new byte[1], 0, -1); } });
- Assert.Throws(null, () => { if (useAsync) { decompressor.WriteAsync(new byte[1], 0, 2).Wait(); } else { decompressor.Write(new byte[1], 0, 2); } });
- Assert.Throws(null, () => { if (useAsync) { decompressor.WriteAsync(new byte[1], 1, 1).Wait(); } else { decompressor.Write(new byte[1], 1, 1); } });
- Assert.Throws(() => useAsync ? decompressor.ReadAsync(new byte[1], 0, 1).Result : decompressor.Read(new byte[1], 0, 1));
- Assert.Throws(null, () => useAsync ? decompressor.ReadAsync(new byte[1], 1, 1).Result : decompressor.Read(new byte[1], 1, 1));
- if (useAsync)
- { decompressor.WriteAsync(new byte[1], 0, 1).Wait(); }
- else
- { decompressor.Write(new byte[1], 0, 1); }
- }
- }
-
- [Theory]
- [InlineData(false)]
- [InlineData(true)]
- public virtual void Read_ArgumentValidation(bool useAsync)
- {
- using (var decompressor = CreateStream(new MemoryStream(), CompressionMode.Decompress))
- {
- Assert.Throws(() => useAsync ? decompressor.ReadAsync(null, 0, 0).Result : decompressor.Read(null, 0, 0));
- Assert.Throws(() => useAsync ? decompressor.ReadAsync(new byte[1], -1, 0).Result : decompressor.Read(new byte[1], -1, 0));
- Assert.Throws(() => useAsync ? decompressor.ReadAsync(new byte[1], 0, -1).Result : decompressor.Read(new byte[1], 0, -1));
- AssertExtensions.Throws(null, () => useAsync ? decompressor.ReadAsync(new byte[1], 0, 2).Result : decompressor.Read(new byte[1], 0, 2));
- AssertExtensions.Throws(null, () => useAsync ? decompressor.ReadAsync(new byte[1], 1, 1).Result : decompressor.Read(new byte[1], 1, 1));
- Assert.Throws(() => { if (useAsync) { decompressor.WriteAsync(new byte[1], 0, 1).Wait(); } else { decompressor.Write(new byte[1], 0, 1); } });
- Assert.Throws(null, () => { if (useAsync) { decompressor.WriteAsync(new byte[1], 1, 1).Wait(); } else { decompressor.Write(new byte[1], 1, 1); } });
-
- var data = new byte[1] { 42 };
- Assert.Equal(0, useAsync ? decompressor.ReadAsync(data, 0, 0).Result : decompressor.Read(data, 0, 0));
- Assert.Equal(42, data[0]);
- }
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void CopyToAsync_ArgumentValidation(CompressionMode mode)
- {
- using (Stream compressor = CreateStream(new MemoryStream(), mode))
- {
- AssertExtensions.Throws("destination", () => { compressor.CopyToAsync(null); });
- AssertExtensions.Throws("bufferSize", () => { compressor.CopyToAsync(new MemoryStream(), 0); });
- Assert.Throws(() => { compressor.CopyToAsync(new MemoryStream(new byte[1], writable: false)); });
- compressor.Dispose();
- Assert.Throws(() => { compressor.CopyToAsync(new MemoryStream()); });
- }
- }
-
- [Theory]
- [InlineData(CompressionMode.Compress)]
- [InlineData(CompressionMode.Decompress)]
- public void CopyTo_ArgumentValidation(CompressionMode mode)
- {
- using (Stream compressor = CreateStream(new MemoryStream(), mode))
- {
- AssertExtensions.Throws("destination", () => { compressor.CopyTo(null); });
- AssertExtensions.Throws("bufferSize", () => { compressor.CopyTo(new MemoryStream(), 0); });
- Assert.Throws(() => { compressor.CopyTo(new MemoryStream(new byte[1], writable: false)); });
- compressor.Dispose();
- Assert.Throws(() => { compressor.CopyTo(new MemoryStream()); });
- }
- }
-
- public enum ReadWriteMode
- {
- SyncArray,
- SyncSpan,
- AsyncArray,
- AsyncMemory,
- AsyncBeginEnd
- }
-
- public static IEnumerable RoundtripCompressDecompressOuterData
- {
- get
- {
- foreach (ReadWriteMode readWriteMode in new[] { ReadWriteMode.SyncArray, ReadWriteMode.SyncSpan, ReadWriteMode.AsyncArray, ReadWriteMode.AsyncMemory, ReadWriteMode.AsyncBeginEnd })
- {
- foreach (var level in new[] { CompressionLevel.Fastest, CompressionLevel.Optimal, CompressionLevel.NoCompression }) // compression level
- {
- yield return new object[] { readWriteMode, 1, 5, level }; // smallest possible writes
- yield return new object[] { readWriteMode, 1023, 1023 * 10, level }; // overflowing internal buffer
- yield return new object[] { readWriteMode, 1024 * 1024, 1024 * 1024, level }; // large single write
- }
- }
- }
- }
-
- [Fact]
- public async Task CompressDecompress_RoundTrip()
- {
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.SyncArray, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.SyncSpan, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.AsyncArray, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.AsyncMemory, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.AsyncBeginEnd, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode.AsyncArray, chunkSize: 1024, totalSize: 8192, level: CompressionLevel.Optimal);
- }
-
- [OuterLoop]
- [Theory]
- [MemberData(nameof(RoundtripCompressDecompressOuterData))]
- public async Task CompressDecompress_RoundTrip_OuterLoop(ReadWriteMode readWriteMode, int chunkSize, int totalSize, CompressionLevel level)
- {
- byte[] data = new byte[totalSize];
- new Random(42).NextBytes(data);
-
- var compressed = new MemoryStream();
- using (var compressor = CreateStream(compressed, level, true))
- {
- for (int i = 0; i < data.Length; i += chunkSize) // not using CopyTo{Async} due to optimizations in MemoryStream's implementation that avoid what we're trying to test
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- await compressor.WriteAsync(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncArray:
- compressor.Write(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncSpan:
- compressor.Write(new ReadOnlySpan(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncMemory:
- await compressor.WriteAsync(new ReadOnlyMemory(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(compressor.BeginWrite, compressor.EndWrite, data, i, chunkSize, null);
- break;
- }
- }
- }
- compressed.Position = 0;
- await ReadAndValidateCompressedData(readWriteMode, chunkSize, compressed, data);
- compressed.Dispose();
- }
-
- [Fact]
- public async Task Flush_RoundTrip()
- {
- if (FlushNoOps)
- return;
- await Flush_RoundTrip_OuterLoop(ReadWriteMode.SyncArray, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await Flush_RoundTrip_OuterLoop(ReadWriteMode.AsyncArray, chunkSize: 1024, totalSize: 8192, level: CompressionLevel.Optimal);
- }
-
- [OuterLoop]
- [Theory]
- [MemberData(nameof(RoundtripCompressDecompressOuterData))]
- public async Task Flush_RoundTrip_OuterLoop(ReadWriteMode readWriteMode, int chunkSize, int totalSize, CompressionLevel level)
- {
- if (FlushNoOps)
- return;
- byte[] data = new byte[totalSize];
- new Random(42).NextBytes(data);
-
- using (var compressed = new MemoryStream())
- using (var compressor = CreateStream(compressed, level, true))
- {
- for (int i = 0; i < data.Length; i += chunkSize) // not using CopyTo{Async} due to optimizations in MemoryStream's implementation that avoid what we're trying to test
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- await compressor.WriteAsync(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncArray:
- compressor.Write(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncSpan:
- compressor.Write(new ReadOnlySpan(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncMemory:
- await compressor.WriteAsync(new ReadOnlyMemory(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(compressor.BeginWrite, compressor.EndWrite, data, i, chunkSize, null);
- break;
- }
- }
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- case ReadWriteMode.AsyncMemory:
- case ReadWriteMode.AsyncBeginEnd:
- await compressor.FlushAsync();
- break;
- case ReadWriteMode.SyncSpan:
- case ReadWriteMode.SyncArray:
- compressor.Flush();
- break;
- }
- if (!FlushCompletes)
- compressor.Dispose();
- compressed.Position = 0;
- await ReadAndValidateCompressedData(readWriteMode, chunkSize, compressed, data);
- }
- }
-
- [Fact]
- public async Task Flush_Consecutive()
- {
- if (FlushNoOps)
- return;
- await Flush_Consecutive_OuterLoop(ReadWriteMode.SyncArray, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await Flush_Consecutive_OuterLoop(ReadWriteMode.AsyncArray, chunkSize: 1024, totalSize: 8192, level: CompressionLevel.Optimal);
- }
-
- [OuterLoop]
- [Theory]
- [MemberData(nameof(RoundtripCompressDecompressOuterData))]
- public async Task Flush_Consecutive_OuterLoop(ReadWriteMode readWriteMode, int chunkSize, int totalSize, CompressionLevel level)
- {
- if (FlushNoOps)
- return;
- byte[] data = new byte[totalSize];
- List expected = new List();
- new Random(42).NextBytes(data);
-
- using (var compressed = new MemoryStream())
- using (var compressor = CreateStream(compressed, level, true))
- {
- for (int i = 0; i < data.Length; i += chunkSize) // not using CopyTo{Async} due to optimizations in MemoryStream's implementation that avoid what we're trying to test
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- await compressor.WriteAsync(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncArray:
- compressor.Write(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncSpan:
- compressor.Write(new ReadOnlySpan(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncMemory:
- await compressor.WriteAsync(new ReadOnlyMemory(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(compressor.BeginWrite, compressor.EndWrite, data, i, chunkSize, null);
- break;
- }
- for (int j = i; j < i + chunkSize; j++)
- expected.Insert(j, data[j]);
-
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- case ReadWriteMode.AsyncMemory:
- case ReadWriteMode.AsyncBeginEnd:
- await compressor.FlushAsync();
- break;
- case ReadWriteMode.SyncSpan:
- case ReadWriteMode.SyncArray:
- compressor.Flush();
- break;
- }
-
- if (FlushCompletes)
- {
- MemoryStream partiallyCompressed = new MemoryStream(compressed.ToArray());
- partiallyCompressed.Position = 0;
- await ReadAndValidateCompressedData(readWriteMode, chunkSize, partiallyCompressed, expected.ToArray());
- }
- }
- if (!FlushCompletes)
- compressor.Dispose();
- MemoryStream fullyCompressed = new MemoryStream(compressed.ToArray());
- fullyCompressed.Position = 0;
- await ReadAndValidateCompressedData(readWriteMode, chunkSize, fullyCompressed, expected.ToArray());
- }
- }
-
- [Fact]
- public async Task Flush_BeforeFirstWrites()
- {
- if (FlushNoOps)
- return;
- await Flush_BeforeFirstWrites_OuterLoop(ReadWriteMode.SyncArray, chunkSize: 1, totalSize: 10, level: CompressionLevel.Fastest);
- await Flush_BeforeFirstWrites_OuterLoop(ReadWriteMode.AsyncArray, chunkSize: 1024, totalSize: 8192, level: CompressionLevel.Optimal);
- }
-
- [OuterLoop]
- [Theory]
- [MemberData(nameof(RoundtripCompressDecompressOuterData))]
- public async Task Flush_BeforeFirstWrites_OuterLoop(ReadWriteMode readWriteMode, int chunkSize, int totalSize, CompressionLevel level)
- {
- if (FlushNoOps)
- return;
- byte[] data = new byte[totalSize];
- new Random(42).NextBytes(data);
-
- using (var compressed = new MemoryStream())
- using (var compressor = CreateStream(compressed, level, true))
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- case ReadWriteMode.AsyncMemory:
- case ReadWriteMode.AsyncBeginEnd:
- await compressor.FlushAsync();
- break;
- case ReadWriteMode.SyncSpan:
- case ReadWriteMode.SyncArray:
- compressor.Flush();
- break;
- }
-
- for (int i = 0; i < data.Length; i += chunkSize) // not using CopyTo{Async} due to optimizations in MemoryStream's implementation that avoid what we're trying to test
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- await compressor.WriteAsync(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncArray:
- compressor.Write(data, i, chunkSize);
- break;
- case ReadWriteMode.SyncSpan:
- compressor.Write(new ReadOnlySpan(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncMemory:
- await compressor.WriteAsync(new ReadOnlyMemory(data, i, chunkSize));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(compressor.BeginWrite, compressor.EndWrite, data, i, chunkSize, null);
- break;
- }
- }
-
- switch (readWriteMode)
- {
- case ReadWriteMode.AsyncArray:
- case ReadWriteMode.AsyncMemory:
- case ReadWriteMode.AsyncBeginEnd:
- await compressor.FlushAsync();
- break;
- case ReadWriteMode.SyncSpan:
- case ReadWriteMode.SyncArray:
- compressor.Flush();
- break;
- }
- if (!FlushCompletes)
- compressor.Dispose();
- compressed.Position = 0;
- await ReadAndValidateCompressedData(readWriteMode, chunkSize, compressed, data);
- }
- }
-
- ///
- /// Given a MemoryStream of compressed data and a byte array of desired output, decompresses
- /// the stream and validates that it is equal to the expected array.
- ///
- private async Task ReadAndValidateCompressedData(ReadWriteMode readWriteMode, int chunkSize, MemoryStream compressed, byte[] expected)
- {
- using (MemoryStream decompressed = new MemoryStream())
- using (Stream decompressor = CreateStream(compressed, CompressionMode.Decompress, true))
- {
- int bytesRead;
- var buffer = new byte[chunkSize];
- switch (readWriteMode)
- {
- case ReadWriteMode.SyncSpan:
- while ((bytesRead = decompressor.Read(new Span(buffer))) != 0)
- {
- decompressed.Write(buffer, 0, bytesRead);
- }
- break;
- case ReadWriteMode.SyncArray:
- while ((bytesRead = decompressor.Read(buffer, 0, buffer.Length)) != 0)
- {
- decompressed.Write(buffer, 0, bytesRead);
- }
- break;
- case ReadWriteMode.AsyncArray:
- while ((bytesRead = await decompressor.ReadAsync(buffer, 0, buffer.Length)) != 0)
- {
- decompressed.Write(buffer, 0, bytesRead);
- }
- break;
- case ReadWriteMode.AsyncMemory:
- while ((bytesRead = await decompressor.ReadAsync(new Memory(buffer))) != 0)
- {
- decompressed.Write(buffer, 0, bytesRead);
- }
- break;
- case ReadWriteMode.AsyncBeginEnd:
- while ((bytesRead = await Task.Factory.FromAsync(decompressor.BeginRead, decompressor.EndRead, buffer, 0, buffer.Length, null)) != 0)
- {
- decompressed.Write(buffer, 0, bytesRead);
- }
- break;
- }
- Assert.Equal(expected, decompressed.ToArray());
- }
- }
-
- [Theory]
- [InlineData(ReadWriteMode.SyncArray)]
- [InlineData(ReadWriteMode.AsyncArray)]
- [InlineData(ReadWriteMode.SyncSpan)]
- [InlineData(ReadWriteMode.AsyncMemory)]
- [InlineData(ReadWriteMode.AsyncBeginEnd)]
- public async Task Read_SequentialReadsOnMemoryStream_Return_SameBytes(ReadWriteMode readWriteMode)
- {
- byte[] data = new byte[1024 * 10];
- new Random(42).NextBytes(data);
-
- var compressed = new MemoryStream();
- using (var compressor = CreateStream(compressed, CompressionMode.Compress, true))
- {
- for (int i = 0; i < data.Length; i += 1024)
- {
- switch (readWriteMode)
- {
- case ReadWriteMode.SyncArray:
- compressor.Write(data, i, 1024);
- break;
- case ReadWriteMode.AsyncArray:
- await compressor.WriteAsync(data, i, 1024);
- break;
- case ReadWriteMode.SyncSpan:
- compressor.Write(new Span(data, i, 1024));
- break;
- case ReadWriteMode.AsyncMemory:
- await compressor.WriteAsync(new ReadOnlyMemory(data, i, 1024));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(compressor.BeginWrite, compressor.EndWrite, data, i, 1024, null);
- break;
- }
- }
- }
- compressed.Position = 0;
-
- using (var decompressor = CreateStream(compressed, CompressionMode.Decompress, true))
- {
- int i, j;
- byte[] array = new byte[100];
- byte[] array2 = new byte[100];
-
- // only read in the first 100 bytes
- switch (readWriteMode)
- {
- case ReadWriteMode.SyncArray:
- decompressor.Read(array, 0, array.Length);
- break;
- case ReadWriteMode.AsyncArray:
- await decompressor.ReadAsync(array, 0, array.Length);
- break;
- case ReadWriteMode.SyncSpan:
- decompressor.Read(new Span(array));
- break;
- case ReadWriteMode.AsyncMemory:
- await decompressor.ReadAsync(new Memory(array));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(decompressor.BeginRead, decompressor.EndRead, array, 0, array.Length, null);
- break;
- }
- for (i = 0; i < array.Length; i++)
- {
- Assert.Equal(data[i], array[i]);
- }
-
- // read in the next 100 bytes and make sure nothing is missing
- switch (readWriteMode)
- {
- case ReadWriteMode.SyncArray:
- decompressor.Read(array2, 0, array2.Length);
- break;
- case ReadWriteMode.AsyncArray:
- await decompressor.ReadAsync(array2, 0, array2.Length);
- break;
- case ReadWriteMode.SyncSpan:
- decompressor.Read(new Span(array2));
- break;
- case ReadWriteMode.AsyncMemory:
- await decompressor.ReadAsync(new Memory(array2));
- break;
- case ReadWriteMode.AsyncBeginEnd:
- await Task.Factory.FromAsync(decompressor.BeginRead, decompressor.EndRead, array2, 0, array2.Length, null);
- break;
- }
- for (j = 0; j < array2.Length; j++)
- {
- Assert.Equal(data[j], array[j]);
- }
- }
- }
-
- [Fact]
- public void WriteByte_RountTrip()
- {
- byte[] data = new byte[1024 * 10];
- new Random(42).NextBytes(data);
-
- var compressed = new MemoryStream();
- using (var compressor = CreateStream(compressed, CompressionMode.Compress, true))
- {
- compressor.Write(data, 0, data.Length);
- }
- compressed.Position = 0;
-
- using (var decompressor = CreateStream(compressed, CompressionMode.Decompress, true))
- {
- for (int i = 0; i < data.Length; i++)
- Assert.Equal(data[i], decompressor.ReadByte());
- }
- }
-
[Fact]
public async Task WrapNullReturningTasksStream()
{
@@ -1100,68 +372,12 @@ public async Task WrapStreamReturningBadReadValues()
using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooLargeCounts), CompressionMode.Decompress))
await Assert.ThrowsAsync(async () => { await decompressor.ReadAsync(new Memory(new byte[1024])); });
- if (!FlushNoOps)
- {
- using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
- Assert.Equal(0, decompressor.Read(new byte[1024], 0, 1024));
- using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
- Assert.Equal(0, await decompressor.ReadAsync(new byte[1024], 0, 1024));
- using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
- Assert.Equal(0, await decompressor.ReadAsync(new Memory(new byte[1024])));
- }
- }
-
- [Theory]
- [OuterLoop]
- [InlineData(true)]
- [InlineData(false)]
- public async Task CopyTo_Roundtrip_OutputMatchesInput(bool useAsync)
- {
- var rand = new Random();
- foreach (int dataSize in new[] { 1, 1024, 4095, 1024 * 1024 })
- {
- var data = new byte[dataSize];
- rand.NextBytes(data);
-
- var compressed = new MemoryStream();
- using (var ds = CreateStream(compressed, CompressionMode.Compress, leaveOpen: true))
- {
- ds.Write(data, 0, data.Length);
- }
- byte[] compressedData = compressed.ToArray();
-
- foreach (int copyBufferSize in new[] { 1, 4096, 80 * 1024 })
- {
- // Memory source
- var m = new MemoryStream(compressedData, writable: false);
- await CopyTo_Roundtrip_OutputMatchesInput_Verify(data, copyBufferSize, m, useAsync);
-
- // File sources, sync and async
- foreach (bool useAsyncFileOptions in new[] { true, false })
- {
- string path = Path.GetTempFileName();
- File.WriteAllBytes(path, compressedData);
-
- FileOptions options = FileOptions.DeleteOnClose;
- if (useAsyncFileOptions)
- options |= FileOptions.Asynchronous;
- await CopyTo_Roundtrip_OutputMatchesInput_Verify(data, copyBufferSize, new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 0x1000, options), useAsync);
- }
- }
- }
- }
-
- private async Task CopyTo_Roundtrip_OutputMatchesInput_Verify(byte[] expectedDecrypted, int copyBufferSize, Stream source, bool useAsync)
- {
- var m = new MemoryStream();
- using (Stream ds = CreateStream(source, CompressionMode.Decompress))
- {
- if (useAsync)
- await ds.CopyToAsync(m);
- else
- ds.CopyTo(m, copyBufferSize);
- }
- Assert.Equal(expectedDecrypted, m.ToArray());
+ using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
+ Assert.Equal(0, decompressor.Read(new byte[1024], 0, 1024));
+ using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
+ Assert.Equal(0, await decompressor.ReadAsync(new byte[1024], 0, 1024));
+ using (var decompressor = CreateStream(new BadWrappedStream(BadWrappedStream.Mode.ReturnTooSmallCounts), CompressionMode.Decompress))
+ Assert.Equal(0, await decompressor.ReadAsync(new Memory(new byte[1024])));
}
[Theory]
@@ -1225,88 +441,6 @@ public async Task BaseStream_ValidAfterDisposeWithTrueLeaveOpen(CompressionMode
else
baseStream.Read(bytes, 0, size);
}
-
- [ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
- public async Task Parallel_CompressDecompressMultipleStreamsConcurrently()
- {
- const int ParallelOperations = 20;
- const int DataSize = 10 * 1024;
-
- var sourceData = new byte[DataSize];
- new Random().NextBytes(sourceData);
-
- await Task.WhenAll(Enumerable.Range(0, ParallelOperations).Select(_ => Task.Run(async () =>
- {
- var compressedStream = new MemoryStream();
- using (Stream ds = CreateStream(compressedStream, CompressionMode.Compress, leaveOpen: true))
- {
- await ds.WriteAsync(sourceData, 0, sourceData.Length);
- }
-
- compressedStream.Position = 0;
-
- var decompressedStream = new MemoryStream();
- using (Stream ds = CreateStream(compressedStream, CompressionMode.Decompress, leaveOpen: true))
- {
- await ds.CopyToAsync(decompressedStream);
- }
-
- Assert.Equal(sourceData, decompressedStream.ToArray());
- })));
- }
-
- [Fact]
- public void Precancellation()
- {
- var ms = new MemoryStream();
- using (Stream compressor = CreateStream(ms, CompressionMode.Compress, leaveOpen: true))
- {
- Assert.True(compressor.WriteAsync(new byte[1], 0, 1, new CancellationToken(true)).IsCanceled);
- Assert.True(compressor.FlushAsync(new CancellationToken(true)).IsCanceled);
- }
- using (Stream decompressor = CreateStream(ms, CompressionMode.Decompress, leaveOpen: true))
- {
- Assert.True(decompressor.ReadAsync(new byte[1], 0, 1, new CancellationToken(true)).IsCanceled);
- }
- }
-
- [Theory]
- [InlineData(false)]
- [InlineData(true)]
- public async Task DisposeAsync_Flushes(bool leaveOpen)
- {
- var ms = new MemoryStream();
- var cs = CreateStream(ms, CompressionMode.Compress, leaveOpen);
- cs.WriteByte(1);
- await cs.FlushAsync();
-
- long pos = ms.Position;
- cs.WriteByte(1);
- Assert.Equal(pos, ms.Position);
-
- await cs.DisposeAsync();
- Assert.InRange(ms.ToArray().Length, pos + 1, int.MaxValue);
- if (leaveOpen)
- {
- Assert.InRange(ms.Position, pos + 1, int.MaxValue);
- }
- else
- {
- Assert.Throws(() => ms.Position);
- }
- }
-
- [Theory]
- [InlineData(false)]
- [InlineData(true)]
- public async Task DisposeAsync_MultipleCallsAllowed(bool leaveOpen)
- {
- using (var cs = CreateStream(new MemoryStream(), CompressionMode.Compress, leaveOpen))
- {
- await cs.DisposeAsync();
- await cs.DisposeAsync();
- }
- }
}
internal sealed class BadWrappedStream : MemoryStream
@@ -1386,7 +520,7 @@ public ManualSyncMemoryStream(bool sync = false) : base()
public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state) => TaskToApm.Begin(WriteAsync(buffer, offset, count), callback, state);
public override void EndWrite(IAsyncResult asyncResult) => TaskToApm.End(asyncResult);
- public override async Task ReadAsync(byte[] array, int offset, int count, CancellationToken cancellationToken)
+ public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
ReadHit = true;
if (isSync)
@@ -1398,10 +532,10 @@ public override async Task ReadAsync(byte[] array, int offset, int count, C
await Task.Run(() => manualResetEvent.Wait(cancellationToken)).ConfigureAwait(false);
}
- return await base.ReadAsync(array, offset, count, cancellationToken);
+ return await base.ReadAsync(buffer, offset, count, cancellationToken);
}
- public override async Task WriteAsync(byte[] array, int offset, int count, CancellationToken cancellationToken)
+ public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
WriteHit = true;
if (isSync)
@@ -1413,7 +547,7 @@ public override async Task WriteAsync(byte[] array, int offset, int count, Cance
await Task.Run(() => manualResetEvent.Wait(cancellationToken)).ConfigureAwait(false);
}
- await base.WriteAsync(array, offset, count, cancellationToken);
+ await base.WriteAsync(buffer, offset, count, cancellationToken);
}
public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken)
diff --git a/src/libraries/Common/tests/System/IO/ConnectedStreams.cs b/src/libraries/Common/tests/System/IO/ConnectedStreams.cs
index a5ea76a245fdab..8d084e2e3cfd60 100644
--- a/src/libraries/Common/tests/System/IO/ConnectedStreams.cs
+++ b/src/libraries/Common/tests/System/IO/ConnectedStreams.cs
@@ -2,10 +2,10 @@
// The .NET Foundation licenses this file to you under the MIT license.
#nullable enable
-using System.IO;
+using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
-using System.Text;
using System.Threading;
using System.Threading.Tasks;
@@ -14,6 +14,36 @@ namespace System.IO
/// Provides support for in-memory producer/consumer streams.
internal sealed class ConnectedStreams
{
+ /// Creates a pair of streams that are connected for unidirectional communication.
+ /// Writing to one stream produces data readable by the either.
+ public static (Stream Writer, Stream Reader) CreateUnidirectional() =>
+ CreateUnidirectional(StreamBuffer.DefaultInitialBufferSize, StreamBuffer.DefaultMaxBufferSize);
+
+ /// Creates a pair of streams that are connected for unidirectional communication.
+ /// The initial buffer size to use when storing data in the connection.
+ /// Writing to one stream produces data readable by the either.
+ public static (Stream Writer, Stream Reader) CreateUnidirectional(int initialBufferSize) =>
+ CreateUnidirectional(initialBufferSize, StreamBuffer.DefaultMaxBufferSize);
+
+ /// Creates a pair of streams that are connected for unidirectional communication.
+ /// The initial buffer size to use when storing data in the connection.
+ ///
+ /// The maximum buffer size to use when storing data in the connection. When this limit is reached,
+ /// writes will block until additional space becomes available.
+ ///
+ /// Writing to one stream produces data readable by the either.
+ public static (Stream Writer, Stream Reader) CreateUnidirectional(int initialBufferSize, int maxBufferSize)
+ {
+ var buffer = new StreamBuffer(initialBufferSize, maxBufferSize);
+
+ // The StreamBuffer is shared between the streams: we don't want to dispose of the underlying storage
+ // in the stream buffer until both streams are done with it, so we ref count and only dispose
+ // when both streams have been disposed. To share the same integer, it's put onto the heap.
+ var refCount = new StrongBox(2);
+
+ return (new UnidirectionalStreamBufferStream(buffer, reader: false, refCount), new UnidirectionalStreamBufferStream(buffer, reader: true, refCount));
+ }
+
/// Creates a pair of streams that are connected for bidirectional communication.
/// Writing to one stream produces data readable by the either, and vice versa.
public static (Stream Stream1, Stream Stream2) CreateBidirectional() =>
@@ -44,17 +74,233 @@ public static (Stream Stream1, Stream Stream2) CreateBidirectional(int initialBu
// when both streams have been disposed. To share the same integer, it's put onto the heap.
var refCount = new StrongBox(2);
- return (new StreamBufferStream(b1, b2, refCount), new StreamBufferStream(b2, b1, refCount));
+ return (new BidirectionalStreamBufferStream(b1, b2, refCount), new BidirectionalStreamBufferStream(b2, b1, refCount));
+ }
+
+ private sealed class UnidirectionalStreamBufferStream : Stream
+ {
+ private readonly StreamBuffer _buffer;
+ private readonly StrongBox _refCount;
+ private readonly bool _reader;
+ private bool _disposed;
+
+ internal UnidirectionalStreamBufferStream(StreamBuffer buffer, bool reader, StrongBox refCount)
+ {
+ _buffer = buffer;
+ _reader = reader;
+ _refCount = refCount;
+ }
+
+ protected override void Dispose(bool disposing)
+ {
+ if (disposing && !_disposed)
+ {
+ _disposed = true;
+
+ if (_reader)
+ {
+ _buffer.AbortRead();
+ }
+ else
+ {
+ _buffer.EndWrite();
+ }
+
+ if (Interlocked.Decrement(ref _refCount.Value) == 0)
+ {
+ _buffer.Dispose();
+ }
+ }
+
+ base.Dispose(disposing);
+ }
+
+ public override bool CanRead => _reader & !_disposed;
+ public override bool CanWrite => !_reader & !_disposed;
+ public override bool CanSeek => false;
+
+ public override void Flush() => ThrowIfDisposed();
+ public override Task FlushAsync(CancellationToken cancellationToken)
+ {
+ ThrowIfDisposed();
+
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled(cancellationToken);
+ }
+
+ return Task.CompletedTask;
+ }
+
+ public override int Read(byte[] buffer, int offset, int count)
+ {
+ ThrowIfDisposed();
+ ValidateArgs(buffer, offset, count);
+ ThrowIfReadingNotSupported();
+
+ return _buffer.Read(new Span(buffer, offset, count));
+ }
+
+ public override int ReadByte()
+ {
+ ThrowIfDisposed();
+ ThrowIfReadingNotSupported();
+
+ byte b = 0;
+ int n = _buffer.Read(MemoryMarshal.CreateSpan(ref b, 1));
+ return n != 0 ? b : -1;
+ }
+
+ public override int Read(Span buffer)
+ {
+ ThrowIfDisposed();
+ ThrowIfReadingNotSupported();
+
+ return _buffer.Read(buffer);
+ }
+
+ public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) =>
+ TaskToApm.Begin(ReadAsync(buffer, offset, count), callback, state);
+
+ public override int EndRead(IAsyncResult asyncResult)
+ {
+ ThrowIfDisposed();
+ ThrowIfReadingNotSupported();
+
+ return TaskToApm.End(asyncResult);
+ }
+
+ public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+ {
+ ThrowIfDisposed();
+ ValidateArgs(buffer, offset, count);
+ ThrowIfReadingNotSupported();
+
+ return _buffer.ReadAsync(new Memory(buffer, offset, count), cancellationToken).AsTask();
+ }
+
+ public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
+ {
+ ThrowIfDisposed();
+ ThrowIfReadingNotSupported();
+
+ return _buffer.ReadAsync(buffer, cancellationToken);
+ }
+
+ public override void Write(byte[] buffer, int offset, int count)
+ {
+ ThrowIfDisposed();
+ ValidateArgs(buffer, offset, count);
+ ThrowIfWritingNotSupported();
+
+ _buffer.Write(new ReadOnlySpan(buffer, offset, count));
+ }
+
+ public override void WriteByte(byte value)
+ {
+ ThrowIfDisposed();
+ ThrowIfWritingNotSupported();
+
+ _buffer.Write(MemoryMarshal.CreateReadOnlySpan(ref value, 1));
+ }
+
+ public override void Write(ReadOnlySpan buffer)
+ {
+ ThrowIfDisposed();
+ ThrowIfWritingNotSupported();
+
+ _buffer.Write(buffer);
+ }
+
+ public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+ {
+ ThrowIfDisposed();
+ ValidateArgs(buffer, offset, count);
+ ThrowIfWritingNotSupported();
+
+ return _buffer.WriteAsync(new ReadOnlyMemory(buffer, offset, count), cancellationToken).AsTask();
+ }
+
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default)
+ {
+ ThrowIfDisposed();
+ ThrowIfWritingNotSupported();
+
+ return _buffer.WriteAsync(buffer, cancellationToken);
+ }
+
+ public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) =>
+ TaskToApm.Begin(WriteAsync(buffer, offset, count), callback, state);
+
+ public override void EndWrite(IAsyncResult asyncResult)
+ {
+ ThrowIfDisposed();
+ ThrowIfWritingNotSupported();
+
+ TaskToApm.End(asyncResult);
+ }
+
+ public override long Length => throw new NotSupportedException();
+ public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
+ public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+ public override void SetLength(long value) => throw new NotSupportedException();
+
+ private void ValidateArgs(byte[] buffer, int offset, int count)
+ {
+ if (buffer is null)
+ {
+ throw new ArgumentNullException(nameof(buffer));
+ }
+
+ if ((uint)offset > buffer.Length)
+ {
+ throw new ArgumentOutOfRangeException(nameof(offset));
+ }
+
+ if ((uint)count > buffer.Length || offset > buffer.Length - count)
+ {
+ throw new ArgumentOutOfRangeException(nameof(count));
+ }
+ }
+
+ private void ThrowIfDisposed()
+ {
+ if (_disposed)
+ ThrowDisposedException();
+
+ [StackTraceHidden]
+ static void ThrowDisposedException() => throw new ObjectDisposedException(nameof(ConnectedStreams));
+ }
+
+ private void ThrowIfReadingNotSupported()
+ {
+ if (!_reader)
+ {
+ ThrowNotSupportedException();
+ }
+ }
+
+ private void ThrowIfWritingNotSupported()
+ {
+ if (_reader)
+ {
+ ThrowNotSupportedException();
+ }
+ }
+
+ [DoesNotReturn]
+ private static void ThrowNotSupportedException() =>
+ throw new NotSupportedException();
}
- private sealed class StreamBufferStream : Stream
+ private sealed class BidirectionalStreamBufferStream : Stream
{
private readonly StreamBuffer _readBuffer;
private readonly StreamBuffer _writeBuffer;
private readonly StrongBox _refCount;
private bool _disposed;
- internal StreamBufferStream(StreamBuffer readBuffer, StreamBuffer writeBuffer, StrongBox refCount)
+ internal BidirectionalStreamBufferStream(StreamBuffer readBuffer, StreamBuffer writeBuffer, StrongBox refCount)
{
_readBuffer = readBuffer;
_writeBuffer = writeBuffer;
@@ -80,67 +326,108 @@ protected override void Dispose(bool disposing)
base.Dispose(disposing);
}
- public override bool CanRead => true;
- public override bool CanWrite => true;
+ public override bool CanRead => !_disposed;
+ public override bool CanWrite => !_disposed;
public override bool CanSeek => false;
- public override void Flush() { }
- public override Task FlushAsync(CancellationToken cancellationToken) =>
- cancellationToken.IsCancellationRequested ? Task.FromCanceled(cancellationToken) :
- Task.CompletedTask;
+ public override void Flush() => ThrowIfDisposed();
+ public override Task FlushAsync(CancellationToken cancellationToken)
+ {
+ ThrowIfDisposed();
+
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled(cancellationToken);
+ }
+
+ return Task.CompletedTask;
+ }
public override int Read(byte[] buffer, int offset, int count)
{
+ ThrowIfDisposed();
ValidateArgs(buffer, offset, count);
+
return _readBuffer.Read(new Span(buffer, offset, count));
}
public override int ReadByte()
{
+ ThrowIfDisposed();
+
byte b = 0;
int n = _readBuffer.Read(MemoryMarshal.CreateSpan(ref b, 1));
return n != 0 ? b : -1;
}
- public override int Read(Span buffer) => _readBuffer.Read(buffer);
+ public override int Read(Span buffer)
+ {
+ ThrowIfDisposed();
+ return _readBuffer.Read(buffer);
+ }
public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) =>
TaskToApm.Begin(ReadAsync(buffer, offset, count), callback, state);
- public override int EndRead(IAsyncResult asyncResult) => TaskToApm.End(asyncResult);
+ public override int EndRead(IAsyncResult asyncResult)
+ {
+ ThrowIfDisposed();
+ return TaskToApm.End(asyncResult);
+ }
public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
+ ThrowIfDisposed();
ValidateArgs(buffer, offset, count);
return _readBuffer.ReadAsync(new Memory(buffer, offset, count), cancellationToken).AsTask();
}
- public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) =>
- _readBuffer.ReadAsync(buffer, cancellationToken);
-
+ public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
+ {
+ ThrowIfDisposed();
+ return _readBuffer.ReadAsync(buffer, cancellationToken);
+ }
public override void Write(byte[] buffer, int offset, int count)
{
+ ThrowIfDisposed();
ValidateArgs(buffer, offset, count);
_writeBuffer.Write(new ReadOnlySpan(buffer, offset, count));
}
- public override void WriteByte(byte value) => _writeBuffer.Write(MemoryMarshal.CreateReadOnlySpan(ref value, 1));
+ public override void WriteByte(byte value)
+ {
+ ThrowIfDisposed();
+ _writeBuffer.Write(MemoryMarshal.CreateReadOnlySpan(ref value, 1));
+ }
- public override void Write(ReadOnlySpan buffer) => _writeBuffer.Write(buffer);
+ public override void Write(ReadOnlySpan buffer)
+ {
+ ThrowIfDisposed();
+ _writeBuffer.Write(buffer);
+ }
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
+ ThrowIfDisposed();
ValidateArgs(buffer, offset, count);
return _writeBuffer.WriteAsync(new ReadOnlyMemory(buffer, offset, count), cancellationToken).AsTask();
}
- public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) => _writeBuffer.WriteAsync(buffer, cancellationToken);
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default)
+ {
+ ThrowIfDisposed();
+ return _writeBuffer.WriteAsync(buffer, cancellationToken);
+ }
public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) =>
TaskToApm.Begin(WriteAsync(buffer, offset, count), callback, state);
- public override void EndWrite(IAsyncResult asyncResult) => TaskToApm.End(asyncResult);
+ public override void EndWrite(IAsyncResult asyncResult)
+ {
+ ThrowIfDisposed();
+ TaskToApm.End(asyncResult);
+ }
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
@@ -164,6 +451,15 @@ private void ValidateArgs(byte[] buffer, int offset, int count)
throw new ArgumentOutOfRangeException(nameof(count));
}
}
+
+ private void ThrowIfDisposed()
+ {
+ if (_disposed)
+ ThrowDisposedException();
+
+ [StackTraceHidden]
+ static void ThrowDisposedException() => throw new ObjectDisposedException(nameof(ConnectedStreams));
+ }
}
}
}
diff --git a/src/libraries/Common/tests/Tests/System/IO/ConnectedStreamsTests.cs b/src/libraries/Common/tests/Tests/System/IO/ConnectedStreamsTests.cs
new file mode 100644
index 00000000000000..0f700bf68f0f71
--- /dev/null
+++ b/src/libraries/Common/tests/Tests/System/IO/ConnectedStreamsTests.cs
@@ -0,0 +1,25 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Threading.Tasks;
+
+namespace System.IO.Tests
+{
+ public class UnidirectionalConnectedStreamsTests : ConnectedStreamConformanceTests
+ {
+ protected override int BufferedSize => StreamBuffer.DefaultMaxBufferSize;
+ protected override bool FlushRequiredToWriteData => false;
+
+ protected override Task CreateConnectedStreamsAsync() =>
+ Task.FromResult(ConnectedStreams.CreateUnidirectional());
+ }
+
+ public class BidirectionalConnectedStreamsTests : ConnectedStreamConformanceTests
+ {
+ protected override int BufferedSize => StreamBuffer.DefaultMaxBufferSize;
+ protected override bool FlushRequiredToWriteData => false;
+
+ protected override Task CreateConnectedStreamsAsync() =>
+ Task.FromResult(ConnectedStreams.CreateBidirectional());
+ }
+}
diff --git a/src/libraries/Common/tests/Tests/System/IO/StreamConformanceTests.cs b/src/libraries/Common/tests/Tests/System/IO/StreamConformanceTests.cs
new file mode 100644
index 00000000000000..9df2a9c3062aac
--- /dev/null
+++ b/src/libraries/Common/tests/Tests/System/IO/StreamConformanceTests.cs
@@ -0,0 +1,1853 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Security.Cryptography;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace System.IO.Tests
+{
+ /// Base class providing tests for any Stream-derived type.
+ [PlatformSpecific(~TestPlatforms.Browser)] // lots of operations aren't supported on browser
+ public abstract class StreamConformanceTests
+ {
+ /// Gets the name of the byte[] argument to Read/Write methods.
+ protected virtual string ReadWriteBufferName => "buffer";
+ /// Gets the name of the int offset argument to Read/Write methods.
+ protected virtual string ReadWriteOffsetName => "offset";
+ /// Gets the name of the int count argument to Read/Write methods.
+ protected virtual string ReadWriteCountName => "count";
+ /// Gets the name of the IAsyncResult argument to EndRead/Write methods.
+ protected virtual string ReadWriteAsyncResultName => "asyncResult";
+ /// Gets the name of the Stream destination argument to CopyTo{Async}.
+ protected virtual string CopyToStreamName => "destination";
+ /// Gets the name of the int bufferSize argument to CopyTo{Async}.
+ protected virtual string CopyToBufferSizeName => "bufferSize";
+
+ /// Gets the type of exception thrown when an invalid IAsyncResult is passed to an EndRead/Write method.
+ protected virtual Type InvalidIAsyncResultExceptionType => typeof(ArgumentException);
+ /// Gets the type of exception thrown when a read or write operation is unsupported.
+ protected virtual Type UnsupportedReadWriteExceptionType => typeof(NotSupportedException);
+ /// Gets the type of exception thrown when a CopyTo{Async} operation is unsupported.
+ protected virtual Type UnsupportedCopyExceptionType => typeof(NotSupportedException);
+ /// Gets the type of exception thrown when setting a Read/WriteTimeout is unsupported.
+ protected virtual Type UnsupportedTimeoutExceptionType => typeof(InvalidOperationException);
+ ///
+ /// Gets the type of exception thrown when an operation is invoked concurrently erroneously, or null if no exception
+ /// is thrown (either because it's fully supported or not supported and non-deterministic).
+ ///
+ protected virtual Type UnsupportedConcurrentExceptionType => typeof(InvalidOperationException);
+
+ /// Gets whether the stream is expected to be seekable.
+ protected virtual bool CanSeek => false;
+ /// Gets whether the stream is expected to support timeouts.
+ protected virtual bool CanTimeout => false;
+ /// Gets whether it's expected for the Position property to be usable even if CanSeek is false.
+ protected virtual bool CanGetPositionWhenCanSeekIsFalse => false;
+ /// Gets whether read/write operations fully support cancellation.
+ protected virtual bool FullyCancelableOperations => true;
+
+ /// Gets whether the stream's CanRead/Write/etc properties are expected to return false once the stream is disposed.
+ protected virtual bool CansReturnFalseAfterDispose => true;
+ /// Gets whether the Stream may be used for additional operations after a read is canceled.
+ protected virtual bool UsableAfterCanceledReads => true;
+
+ /// Specifies the form of the read/write operation to use.
+ public enum ReadWriteMode
+ {
+ /// ReadByte / WriteByte
+ SyncByte,
+ /// Read(Span{byte}) / Write(ReadOnlySpan{byte})
+ SyncSpan,
+ /// Read(byte[], int, int) / Write(byte[], int, int)
+ SyncArray,
+ /// ReadAsync(byte[], int, int) / WriteAsync(byte[], int, int)
+ AsyncArray,
+ /// ReadAsync(Memory{byte}) / WriteAsync(ReadOnlyMemory{byte})
+ AsyncMemory,
+ /// EndRead(BeginRead(..., null, null)) / EndWrite(BeginWrite(..., null, null))
+ SyncAPM,
+ /// Task.Factory.FromAsync(s.BeginRead, s.EndRead, ...) / Task.Factory.FromAsync(s.BeginWrite, s.EndWrite, ...)
+ AsyncAPM
+ }
+
+ protected async Task ValidateMisuseExceptionsAsync(Stream stream)
+ {
+ byte[] oneByteBuffer = new byte[1];
+
+ if (stream.CanRead)
+ {
+ // Null arguments
+ foreach ((int offset, int count) in new[] { (0, 0), (1, 2) }) // validate 0, 0 isn't special-cased to be allowed with a null buffer
+ {
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.Read(null, offset, count); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.Read(null, offset, count); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.ReadAsync(null, offset, count); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.ReadAsync(null, offset, count, default); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.EndRead(stream.BeginRead(null, offset, count, iar => { }, new object())); });
+ AssertExtensions.Throws(ReadWriteAsyncResultName, () => { stream.EndRead(null); });
+ }
+
+ // Invalid offset
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.Read(oneByteBuffer, -1, 0); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.ReadAsync(oneByteBuffer, -1, 0); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.ReadAsync(oneByteBuffer, -1, 0, default); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.EndRead(stream.BeginRead(oneByteBuffer, -1, 0, iar => { }, new object())); });
+ Assert.ThrowsAny(() => { stream.Read(oneByteBuffer, 2, 0); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, 2, 0); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, 2, 0, default); });
+ Assert.ThrowsAny(() => { stream.EndRead(stream.BeginRead(oneByteBuffer, 2, 0, iar => { }, new object())); });
+
+ // Invalid count
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.Read(oneByteBuffer, 0, -1); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.ReadAsync(oneByteBuffer, 0, -1); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.ReadAsync(oneByteBuffer, 0, -1, default); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.EndRead(stream.BeginRead(oneByteBuffer, 0, -1, iar => { }, new object())); });
+ Assert.ThrowsAny(() => { stream.Read(oneByteBuffer, 0, 2); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, 0, 2); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, 0, 2, default); });
+ Assert.ThrowsAny(() => { stream.EndRead(stream.BeginRead(oneByteBuffer, 0, 2, iar => { }, new object())); });
+
+ // Invalid offset + count
+ foreach ((int invalidOffset, int invalidCount) in new[] { (1, 1) })
+ {
+ Assert.ThrowsAny(() => { stream.Read(oneByteBuffer, invalidOffset, invalidCount); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, invalidOffset, invalidCount); });
+ Assert.ThrowsAny(() => { stream.ReadAsync(oneByteBuffer, invalidOffset, invalidCount, default); });
+ Assert.ThrowsAny(() => { stream.EndRead(stream.BeginRead(oneByteBuffer, invalidOffset, invalidCount, iar => { }, new object())); });
+ }
+
+ // Unknown arguments
+ Assert.Throws(InvalidIAsyncResultExceptionType, () => stream.EndRead(new NotImplementedIAsyncResult()));
+
+ // Invalid destination stream
+ AssertExtensions.Throws(CopyToStreamName, () => { stream.CopyTo(null); });
+ AssertExtensions.Throws(CopyToStreamName, () => { stream.CopyTo(null, 1); });
+ AssertExtensions.Throws(CopyToStreamName, () => { stream.CopyToAsync(null, default(CancellationToken)); });
+ AssertExtensions.Throws(CopyToStreamName, () => { stream.CopyToAsync(null, 1); });
+ AssertExtensions.Throws(CopyToStreamName, () => { stream.CopyToAsync(null, 1, default(CancellationToken)); });
+
+ // Invalid buffer size
+ var validDestinationStream = new MemoryStream();
+ foreach (int invalidBufferSize in new[] { 0, -1 })
+ {
+ AssertExtensions.Throws(CopyToBufferSizeName, () => { stream.CopyTo(validDestinationStream, invalidBufferSize); });
+ AssertExtensions.Throws(CopyToBufferSizeName, () => { stream.CopyToAsync(validDestinationStream, invalidBufferSize); });
+ AssertExtensions.Throws(CopyToBufferSizeName, () => { stream.CopyToAsync(validDestinationStream, invalidBufferSize, default(CancellationToken)); });
+ }
+
+ // Unwriteable destination stream
+ var unwriteableDestination = new MemoryStream(new byte[1], writable: false);
+ Assert.Throws(UnsupportedCopyExceptionType, () => { stream.CopyTo(unwriteableDestination); });
+ Assert.Throws(UnsupportedCopyExceptionType, () => { stream.CopyToAsync(unwriteableDestination); });
+
+ // Disposed destination stream
+ var disposedDestination = new MemoryStream(new byte[1]);
+ disposedDestination.Dispose();
+ Assert.Throws(() => { stream.CopyTo(disposedDestination); });
+ Assert.Throws(() => { stream.CopyToAsync(disposedDestination); });
+ }
+ else
+ {
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.ReadByte(); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.Read(new Span(new byte[1])); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.Read(new byte[1], 0, 1); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.ReadAsync(new byte[1], 0, 1); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.ReadAsync(new Memory(new byte[1])); });
+ await Assert.ThrowsAsync(UnsupportedReadWriteExceptionType, () => Task.Factory.FromAsync(stream.BeginRead, stream.EndRead, new byte[1], 0, 1, null));
+ Assert.True(Record.Exception(() => stream.EndRead(new NotImplementedIAsyncResult())) is Exception e && (e.GetType() == UnsupportedReadWriteExceptionType || e.GetType() == InvalidIAsyncResultExceptionType));
+ Assert.Throws(UnsupportedCopyExceptionType, () => { stream.CopyTo(new MemoryStream()); });
+ Assert.Throws(UnsupportedCopyExceptionType, () => { stream.CopyToAsync(new MemoryStream()); });
+ }
+
+ if (stream.CanWrite)
+ {
+ // Null arguments
+ foreach ((int offset, int count) in new[] { (0, 0), (1, 2) }) // validate 0, 0 isn't special-cased to be allowed with a null buffer
+ {
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.Write(null, offset, count); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.WriteAsync(null, offset, count); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.WriteAsync(null, offset, count, default); });
+ AssertExtensions.Throws(ReadWriteBufferName, () => { stream.EndWrite(stream.BeginWrite(null, offset, count, iar => { }, new object())); });
+ AssertExtensions.Throws(ReadWriteAsyncResultName, () => { stream.EndWrite(null); });
+ }
+
+ // Invalid offset
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.Write(oneByteBuffer, -1, 0); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.WriteAsync(oneByteBuffer, -1, 0); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.WriteAsync(oneByteBuffer, -1, 0, default); });
+ AssertExtensions.Throws(ReadWriteOffsetName, () => { stream.EndWrite(stream.BeginWrite(oneByteBuffer, -1, 0, iar => { }, new object())); });
+ Assert.ThrowsAny(() => { stream.Write(oneByteBuffer, 2, 0); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, 2, 0); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, 2, 0, default); });
+ Assert.ThrowsAny(() => { stream.EndWrite(stream.BeginWrite(oneByteBuffer, 2, 0, iar => { }, new object())); });
+
+ // Invalid count
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.Write(oneByteBuffer, 0, -1); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.WriteAsync(oneByteBuffer, 0, -1); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.WriteAsync(oneByteBuffer, 0, -1, default); });
+ AssertExtensions.Throws(ReadWriteCountName, () => { stream.EndWrite(stream.BeginWrite(oneByteBuffer, 0, -1, iar => { }, new object())); });
+ Assert.ThrowsAny(() => { stream.Write(oneByteBuffer, 0, 2); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, 0, 2); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, 0, 2, default); });
+ Assert.ThrowsAny(() => { stream.EndWrite(stream.BeginWrite(oneByteBuffer, 0, 2, iar => { }, new object())); });
+
+ // Invalid offset + count
+ foreach ((int invalidOffset, int invalidCount) in new[] { (1, 1) })
+ {
+ Assert.ThrowsAny(() => { stream.Write(oneByteBuffer, invalidOffset, invalidCount); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, invalidOffset, invalidCount); });
+ Assert.ThrowsAny(() => { stream.WriteAsync(oneByteBuffer, invalidOffset, invalidCount, default); });
+ Assert.ThrowsAny(() => { stream.EndWrite(stream.BeginWrite(oneByteBuffer, invalidOffset, invalidCount, iar => { }, new object())); });
+ }
+
+ // Unknown arguments
+ Assert.Throws(InvalidIAsyncResultExceptionType, () => stream.EndWrite(new NotImplementedIAsyncResult()));
+ }
+ else
+ {
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.WriteByte(1); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.Write(new Span(new byte[1])); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.Write(new byte[1], 0, 1); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.WriteAsync(new byte[1], 0, 1); });
+ Assert.Throws(UnsupportedReadWriteExceptionType, () => { stream.WriteAsync(new Memory(new byte[1])); });
+ await Assert.ThrowsAsync(UnsupportedReadWriteExceptionType, () => Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, new byte[1], 0, 1, null));
+ Assert.True(Record.Exception(() => stream.EndWrite(new NotImplementedIAsyncResult())) is Exception e && (e.GetType() == UnsupportedReadWriteExceptionType || e.GetType() == InvalidIAsyncResultExceptionType));
+ }
+
+ Assert.Equal(CanSeek, stream.CanSeek);
+ if (stream.CanSeek)
+ {
+ Assert.Throws(() => { stream.Position = -1; });
+ Assert.Throws(() => { stream.Seek(-1, SeekOrigin.Begin); });
+ Assert.Throws(() => { stream.Seek(0, (SeekOrigin)(-1)); });
+ Assert.Throws(() => { stream.Seek(0, (SeekOrigin)3); });
+ Assert.Throws(() => { stream.SetLength(-1); });
+ }
+ else
+ {
+ Assert.Throws(() => stream.Length);
+ if (!CanGetPositionWhenCanSeekIsFalse)
+ {
+ Assert.Throws(() => stream.Position);
+ }
+ Assert.Throws(() => { stream.Position = 0; });
+ Assert.Throws(() => { stream.SetLength(1); });
+ Assert.Throws(() => { stream.Seek(0, SeekOrigin.Begin); });
+ }
+
+ Assert.Equal(CanTimeout, stream.CanTimeout);
+ if (stream.CanTimeout)
+ {
+ Assert.Throws(() => stream.ReadTimeout = 0);
+ Assert.Throws(() => stream.ReadTimeout = -2);
+ Assert.Throws(() => stream.WriteTimeout = 0);
+ Assert.Throws(() => stream.WriteTimeout = -2);
+ }
+ else
+ {
+ Assert.Throws(UnsupportedTimeoutExceptionType, () => stream.ReadTimeout);
+ Assert.Throws(UnsupportedTimeoutExceptionType, () => stream.ReadTimeout = 1);
+ Assert.Throws(UnsupportedTimeoutExceptionType, () => stream.WriteTimeout);
+ Assert.Throws(UnsupportedTimeoutExceptionType, () => stream.WriteTimeout = 1);
+ }
+ }
+
+ protected async Task ValidateDisposedExceptionsAsync(Stream stream)
+ {
+ // Disposal should be idempotent and not throw
+ stream.Dispose();
+ stream.DisposeAsync().AsTask().GetAwaiter().GetResult();
+ stream.Close();
+
+ AssertDisposed(() => { stream.ReadByte(); });
+ AssertDisposed(() => { stream.Read(new Span(new byte[1])); });
+ AssertDisposed(() => { stream.Read(new byte[1], 0, 1); });
+ await AssertDisposedAsync(async () => await stream.ReadAsync(new byte[1], 0, 1));
+ await AssertDisposedAsync(async() => await stream.ReadAsync(new Memory(new byte[1])));
+ AssertDisposed(() => { stream.EndRead(stream.BeginRead(new byte[1], 0, 1, null, null)); });
+
+ AssertDisposed(() => { stream.WriteByte(1); });
+ AssertDisposed(() => { stream.Write(new Span(new byte[1])); });
+ AssertDisposed(() => { stream.Write(new byte[1], 0, 1); });
+ await AssertDisposedAsync(async () => await stream.WriteAsync(new byte[1], 0, 1));
+ await AssertDisposedAsync(async() => await stream.WriteAsync(new Memory(new byte[1])));
+ AssertDisposed(() => { stream.EndWrite(stream.BeginWrite(new byte[1], 0, 1, null, null)); });
+
+ AssertDisposed(() => stream.Flush(), successAllowed: true);
+ await AssertDisposedAsync(() => stream.FlushAsync(), successAllowed: true);
+
+ AssertDisposed(() => { stream.CopyTo(new MemoryStream()); });
+ await AssertDisposedAsync(async () => await stream.CopyToAsync(new MemoryStream()));
+
+ AssertDisposed(() => _ = stream.Length);
+ AssertDisposed(() => _ = stream.Position);
+ AssertDisposed(() => stream.Position = 0);
+ AssertDisposed(() => stream.Seek(0, SeekOrigin.Begin));
+ AssertDisposed(() => stream.SetLength(1));
+
+ AssertDisposed(() => _ = stream.ReadTimeout);
+ AssertDisposed(() => stream.ReadTimeout = 1);
+ AssertDisposed(() => _ = stream.WriteTimeout);
+ AssertDisposed(() => stream.WriteTimeout = 1);
+
+ void AssertDisposed(Action action, bool successAllowed = false) => ValidateDisposedException(Record.Exception(action), successAllowed);
+
+ async Task AssertDisposedAsync(Func func, bool successAllowed = false) => ValidateDisposedException(await Record.ExceptionAsync(func).ConfigureAwait(false), successAllowed);
+
+ void ValidateDisposedException(Exception e, bool successAllowed = false)
+ {
+ // Behavior when disposed is inconsistent, and isn't specified by the Stream contract: types aren't supposed to be used
+ // after they're disposed. So, at least until we decide to be more strict, these tests are very liberal in what they except.
+ Assert.True(
+ (e is null && successAllowed) ||
+ e is ObjectDisposedException ||
+ e is NotSupportedException ||
+ e is InvalidOperationException,
+ $"Unexpected: {e?.GetType().ToString() ?? "(null)"}");
+ }
+ }
+
+ protected async Task ValidatePrecanceledOperations_ThrowsCancellationException(Stream stream)
+ {
+ var cts = new CancellationTokenSource();
+ cts.Cancel();
+
+ OperationCanceledException oce;
+
+ if (stream.CanRead)
+ {
+ oce = await Assert.ThrowsAnyAsync(() => stream.ReadAsync(new byte[1], 0, 1, cts.Token));
+ Assert.Equal(cts.Token, oce.CancellationToken);
+
+ oce = await Assert.ThrowsAnyAsync(async () => { await stream.ReadAsync(new Memory(new byte[1]), cts.Token); });
+ Assert.Equal(cts.Token, oce.CancellationToken);
+ }
+
+ if (stream.CanWrite)
+ {
+ oce = await Assert.ThrowsAnyAsync(() => stream.WriteAsync(new byte[1], 0, 1, cts.Token));
+ Assert.Equal(cts.Token, oce.CancellationToken);
+
+ oce = await Assert.ThrowsAnyAsync(async () => { await stream.WriteAsync(new ReadOnlyMemory(new byte[1]), cts.Token); });
+ Assert.Equal(cts.Token, oce.CancellationToken);
+ }
+
+ Exception e = await Record.ExceptionAsync(() => stream.FlushAsync(cts.Token));
+ if (e != null)
+ {
+ Assert.IsAssignableFrom(e);
+ }
+ }
+
+ protected async Task ValidateCancelableReads_AfterInvocation_ThrowsCancellationException(Stream stream)
+ {
+ if (!stream.CanRead || !FullyCancelableOperations)
+ {
+ return;
+ }
+
+ CancellationTokenSource cts;
+ OperationCanceledException oce;
+
+ cts = new CancellationTokenSource(1);
+ oce = await Assert.ThrowsAnyAsync(() => stream.ReadAsync(new byte[1], 0, 1, cts.Token));
+ Assert.Equal(cts.Token, oce.CancellationToken);
+
+ cts = new CancellationTokenSource(1);
+ oce = await Assert.ThrowsAnyAsync(async () => { await stream.ReadAsync(new Memory(new byte[1]), cts.Token); });
+ Assert.Equal(cts.Token, oce.CancellationToken);
+ }
+
+ protected async Task WhenAllOrAnyFailed(Task task1, Task task2)
+ {
+ Task completed = await Task.WhenAny(task1, task2);
+ Task incomplete = task1 == completed ? task2 : task1;
+ if (completed.IsCompletedSuccessfully)
+ {
+ await incomplete;
+ }
+ else
+ {
+ var cts = new CancellationTokenSource();
+ await Task.WhenAny(incomplete, Task.Delay(500, cts.Token)); // give second task a chance to complete
+ cts.Cancel();
+ await (incomplete.IsCompleted ? Task.WhenAll(completed, incomplete) : completed);
+ }
+ }
+
+ protected sealed class NotImplementedIAsyncResult : IAsyncResult
+ {
+ public object AsyncState => throw new NotImplementedException();
+ public WaitHandle AsyncWaitHandle => throw new NotImplementedException();
+ public bool CompletedSynchronously => throw new NotImplementedException();
+ public bool IsCompleted => throw new NotImplementedException();
+ }
+
+ protected sealed class CustomSynchronizationContext : SynchronizationContext
+ {
+ public override void Post(SendOrPostCallback d, object state)
+ {
+ ThreadPool.QueueUserWorkItem(delegate
+ {
+ SetSynchronizationContext(this);
+ try
+ {
+ d(state);
+ }
+ finally
+ {
+ SetSynchronizationContext(null);
+ }
+ }, null);
+ }
+ }
+
+ protected sealed class CustomTaskScheduler : TaskScheduler
+ {
+ protected override void QueueTask(Task task) => ThreadPool.QueueUserWorkItem(_ => TryExecuteTask(task));
+ protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) => false;
+ protected override IEnumerable GetScheduledTasks() => null;
+ }
+
+ protected readonly struct JumpToThreadPoolAwaiter : ICriticalNotifyCompletion
+ {
+ public JumpToThreadPoolAwaiter GetAwaiter() => this;
+ public bool IsCompleted => false;
+ public void OnCompleted(Action continuation) => ThreadPool.QueueUserWorkItem(_ => continuation());
+ public void UnsafeOnCompleted(Action continuation) => ThreadPool.UnsafeQueueUserWorkItem(_ => continuation(), null);
+ public void GetResult() { }
+ }
+
+ protected sealed class MisbehavingDelegatingStream : Stream
+ {
+ public enum Mode
+ {
+ Default,
+ ReturnNullTasks,
+ ReturnTooSmallCounts,
+ ReturnTooLargeCounts,
+ ReadSlowly
+ }
+
+ private readonly Stream _stream;
+ private readonly Mode _mode;
+
+ public MisbehavingDelegatingStream(Stream innerStream, Mode mode)
+ {
+ _stream = innerStream;
+ _mode = mode;
+ }
+
+ public override int Read(byte[] buffer, int offset, int count)
+ {
+ switch (_mode)
+ {
+ case Mode.ReturnTooSmallCounts:
+ return -1;
+ case Mode.ReturnTooLargeCounts:
+ return buffer.Length + 1;
+ case Mode.ReadSlowly:
+ return _stream.Read(buffer, offset, 1);
+ default:
+ return 0;
+ }
+ }
+
+ public override void Write(byte[] buffer, int offset, int count) =>
+ _stream.Write(buffer, offset, count);
+
+ public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
+ _mode == Mode.ReturnNullTasks ?
+ null :
+ base.ReadAsync(buffer, offset, count, cancellationToken);
+
+ public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
+ _mode == Mode.ReturnNullTasks ?
+ null :
+ base.WriteAsync(buffer, offset, count, cancellationToken);
+
+ public override void Flush() => _stream.Flush();
+ public override bool CanRead => _stream.CanRead;
+ public override bool CanSeek => _stream.CanSeek;
+ public override bool CanWrite => _stream.CanWrite;
+ public override long Length => _stream.Length;
+ public override long Position { get => _stream.Position; set => _stream.Position = value; }
+ public override long Seek(long offset, SeekOrigin origin) => _stream.Seek(offset, origin);
+ public override void SetLength(long value) => _stream.SetLength(value);
+ }
+ }
+
+ /// Base class providing tests for two streams connected to each other such that writing to one is readable from the other, and vice versa.
+ public abstract class ConnectedStreamConformanceTests : StreamConformanceTests
+ {
+ /// Gets whether ValueTasks returned from Read/WriteAsync methods are expected to be consumable only once.
+ protected virtual bool ReadWriteValueTasksProtectSingleConsumption => false;
+ /// Gets whether writes on a connected stream are expected to fail immediately after a reader is disposed.
+ protected virtual bool BrokenPipePropagatedImmediately => false;
+ /// Gets the amount of data a writer is able to buffer before blocking subsequent writes, or -1 if there's no such limit known.
+ protected virtual int BufferedSize => -1;
+ ///
+ /// Gets whether the stream requires Flush{Async} to be called in order to send written data to the underlying destination.
+ ///
+ protected virtual bool FlushRequiredToWriteData => true;
+ ///
+ /// Gets whether the stream guarantees that all data written to it will be flushed as part of Flush{Async}.
+ ///
+ protected virtual bool FlushGuaranteesAllDataWritten => true;
+ ///
+ /// Gets whether a stream implements an aggressive read that tries to fill the supplied buffer and only
+ /// stops when it does so or hits EOF.
+ ///
+ protected virtual bool ReadsMayBlockUntilBufferFullOrEOF => false;
+ /// Gets whether reads for a count of 0 bytes block if no bytes are available to read.
+ protected virtual bool BlocksOnZeroByteReads => false;
+ ///
+ /// Gets whether an otherwise bidirectional stream does not support reading/writing concurrently, e.g. due to a semaphore in the base implementation.
+ ///
+ protected virtual bool SupportsConcurrentBidirectionalUse => true;
+
+ protected abstract Task CreateConnectedStreamsAsync();
+
+ protected (Stream writeable, Stream readable) GetReadWritePair(StreamPair streams) =>
+ GetReadWritePairs(streams).First();
+
+ protected IEnumerable<(Stream writeable, Stream readable)> GetReadWritePairs(StreamPair streams)
+ {
+ var pairs = new List<(Stream, Stream)>(2);
+
+ if (streams.Stream1.CanWrite)
+ {
+ Assert.True(streams.Stream2.CanRead);
+ pairs.Add((streams.Stream1, streams.Stream2));
+ }
+
+ if (streams.Stream2.CanWrite)
+ {
+ Assert.True(streams.Stream1.CanRead);
+ pairs.Add((streams.Stream2, streams.Stream1));
+ }
+
+ Assert.InRange(pairs.Count, 1, 2);
+ return pairs;
+ }
+
+ protected static bool Bidirectional(StreamPair streams) =>
+ streams.Stream1.CanRead && streams.Stream1.CanWrite &&
+ streams.Stream2.CanRead && streams.Stream2.CanWrite;
+
+ [Fact]
+ public virtual async Task ArgumentValidation_ThrowsExpectedException()
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+
+ foreach (Stream stream in streams)
+ {
+ await ValidateMisuseExceptionsAsync(stream);
+ }
+ }
+
+ [Fact]
+ public virtual async Task Disposed_ThrowsObjectDisposedException()
+ {
+ StreamPair streams = await CreateConnectedStreamsAsync();
+ streams.Dispose();
+
+ foreach (Stream stream in streams)
+ {
+ await ValidateDisposedExceptionsAsync(stream);
+ }
+ }
+
+ [Fact]
+ public virtual async Task ReadWriteAsync_Canceled_ThrowsOperationCanceledException()
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+
+ foreach (Stream stream in streams)
+ {
+ await ValidatePrecanceledOperations_ThrowsCancellationException(stream);
+ await ValidateCancelableReads_AfterInvocation_ThrowsCancellationException(stream);
+ }
+ }
+
+ [Fact]
+ public virtual async Task ReadWriteByte_Success()
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ byte[] writerBytes = RandomNumberGenerator.GetBytes(42);
+ var readerBytes = new byte[writerBytes.Length];
+
+ Task writes = Task.Run(() =>
+ {
+ foreach (byte b in writerBytes)
+ {
+ writeable.WriteByte(b);
+ }
+
+ if (FlushRequiredToWriteData)
+ {
+ if (FlushGuaranteesAllDataWritten)
+ {
+ writeable.Flush();
+ }
+ else
+ {
+ writeable.Dispose();
+ }
+ }
+ });
+
+ for (int i = 0; i < readerBytes.Length; i++)
+ {
+ int r = readable.ReadByte();
+ Assert.InRange(r, 0, 255);
+ readerBytes[i] = (byte)r;
+ }
+
+ Assert.Equal(writerBytes, readerBytes);
+
+ await writes;
+
+ if (!FlushGuaranteesAllDataWritten)
+ {
+ break;
+ }
+ }
+ }
+
+ public static IEnumerable ReadWrite_Success_MemberData() =>
+ from mode in new[] { ReadWriteMode.SyncSpan, ReadWriteMode.SyncArray, ReadWriteMode.SyncAPM, ReadWriteMode.AsyncArray, ReadWriteMode.AsyncMemory, ReadWriteMode.AsyncAPM }
+ from writeSize in new[] { 1, 42, 10 * 1024 }
+ from startWithFlush in new[] { false, true }
+ select new object[] { mode, writeSize, startWithFlush };
+
+ public static IEnumerable ReadWrite_Success_Large_MemberData() =>
+ from mode in new[] { ReadWriteMode.SyncSpan, ReadWriteMode.SyncArray, ReadWriteMode.SyncAPM, ReadWriteMode.AsyncArray, ReadWriteMode.AsyncMemory, ReadWriteMode.AsyncAPM }
+ from writeSize in new[] { 10 * 1024 * 1024 }
+ from startWithFlush in new[] { false, true }
+ select new object[] { mode, writeSize, startWithFlush };
+
+ [OuterLoop]
+ [Theory]
+ [MemberData(nameof(ReadWrite_Success_Large_MemberData))]
+ public virtual async Task ReadWrite_Success_Large(ReadWriteMode mode, int writeSize, bool startWithFlush) =>
+ await ReadWrite_Success(mode, writeSize, startWithFlush);
+
+ [Theory]
+ [MemberData(nameof(ReadWrite_Success_MemberData))]
+ public virtual async Task ReadWrite_Success(ReadWriteMode mode, int writeSize, bool startWithFlush)
+ {
+ foreach (CancellationToken nonCanceledToken in new[] { CancellationToken.None, new CancellationTokenSource().Token })
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ if (startWithFlush)
+ {
+ switch (mode)
+ {
+ case ReadWriteMode.SyncArray:
+ case ReadWriteMode.SyncSpan:
+ case ReadWriteMode.SyncAPM:
+ writeable.Flush();
+ break;
+
+ case ReadWriteMode.AsyncArray:
+ case ReadWriteMode.AsyncMemory:
+ case ReadWriteMode.AsyncAPM:
+ await writeable.FlushAsync(nonCanceledToken);
+ break;
+
+ default:
+ throw new Exception($"Unknown mode: {mode}");
+ }
+ }
+
+ byte[] writerBytes = RandomNumberGenerator.GetBytes(writeSize);
+ var readerBytes = new byte[writerBytes.Length];
+
+ Task writes = Task.Run(async () =>
+ {
+ switch (mode)
+ {
+ case ReadWriteMode.SyncArray:
+ writeable.Write(writerBytes, 0, writerBytes.Length);
+ break;
+
+ case ReadWriteMode.SyncSpan:
+ writeable.Write(writerBytes);
+ break;
+
+ case ReadWriteMode.AsyncArray:
+ await writeable.WriteAsync(writerBytes, 0, writerBytes.Length, nonCanceledToken);
+ break;
+
+ case ReadWriteMode.AsyncMemory:
+ await writeable.WriteAsync(writerBytes, nonCanceledToken);
+ break;
+
+ case ReadWriteMode.SyncAPM:
+ writeable.EndWrite(writeable.BeginWrite(writerBytes, 0, writerBytes.Length, null, null));
+ break;
+
+ case ReadWriteMode.AsyncAPM:
+ await Task.Factory.FromAsync(writeable.BeginWrite, writeable.EndWrite, writerBytes, 0, writerBytes.Length, null);
+ break;
+
+ default:
+ throw new Exception($"Unknown mode: {mode}");
+ }
+
+ if (FlushRequiredToWriteData)
+ {
+ if (FlushGuaranteesAllDataWritten)
+ {
+ await writeable.FlushAsync();
+ }
+ else
+ {
+ await writeable.DisposeAsync();
+ }
+ }
+ });
+
+ int n = 0;
+ while (n < readerBytes.Length)
+ {
+ int r = mode switch
+ {
+ ReadWriteMode.SyncArray => readable.Read(readerBytes, n, readerBytes.Length - n),
+ ReadWriteMode.SyncSpan => readable.Read(readerBytes.AsSpan(n)),
+ ReadWriteMode.AsyncArray => await readable.ReadAsync(readerBytes, n, readerBytes.Length - n, nonCanceledToken),
+ ReadWriteMode.AsyncMemory => await readable.ReadAsync(readerBytes.AsMemory(n), nonCanceledToken),
+ ReadWriteMode.SyncAPM => readable.EndRead(readable.BeginRead(readerBytes, n, readerBytes.Length - n, null, null)),
+ ReadWriteMode.AsyncAPM => await Task.Factory.FromAsync(readable.BeginRead, readable.EndRead, readerBytes, n, readerBytes.Length - n, null),
+ _ => throw new Exception($"Unknown mode: {mode}"),
+ };
+ Assert.InRange(r, 1, readerBytes.Length - n);
+ n += r;
+ }
+
+ Assert.Equal(readerBytes.Length, n);
+ Assert.Equal(writerBytes, readerBytes);
+
+ await writes;
+
+ if (!FlushGuaranteesAllDataWritten)
+ {
+ break;
+ }
+ }
+ }
+ }
+
+ [Theory]
+ [InlineData(ReadWriteMode.SyncByte, false)]
+ [InlineData(ReadWriteMode.SyncArray, false)]
+ [InlineData(ReadWriteMode.SyncSpan, false)]
+ [InlineData(ReadWriteMode.AsyncArray, false)]
+ [InlineData(ReadWriteMode.AsyncMemory, false)]
+ [InlineData(ReadWriteMode.SyncAPM, false)]
+ [InlineData(ReadWriteMode.AsyncAPM, false)]
+ [InlineData(ReadWriteMode.SyncByte, true)]
+ [InlineData(ReadWriteMode.SyncArray, true)]
+ [InlineData(ReadWriteMode.SyncSpan, true)]
+ [InlineData(ReadWriteMode.AsyncArray, true)]
+ [InlineData(ReadWriteMode.AsyncMemory, true)]
+ [InlineData(ReadWriteMode.SyncAPM, true)]
+ [InlineData(ReadWriteMode.AsyncAPM, true)]
+ public virtual async Task Read_Eof_Returns0(ReadWriteMode mode, bool dataAvailableFirst)
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ (Stream writeable, Stream readable) = GetReadWritePair(streams);
+
+ Task write;
+ if (dataAvailableFirst)
+ {
+ write = Task.Run(async () =>
+ {
+ await writeable.WriteAsync(Encoding.UTF8.GetBytes("hello"));
+ await writeable.DisposeAsync();
+ });
+ }
+ else
+ {
+ writeable.Dispose();
+ write = Task.CompletedTask;
+ }
+
+ if (dataAvailableFirst)
+ {
+ Assert.Equal('h', readable.ReadByte());
+ Assert.Equal('e', readable.ReadByte());
+ Assert.Equal('l', readable.ReadByte());
+ Assert.Equal('l', readable.ReadByte());
+ Assert.Equal('o', readable.ReadByte());
+ }
+
+ await write;
+
+ if (mode == ReadWriteMode.SyncByte)
+ {
+ Assert.Equal(-1, readable.ReadByte());
+ }
+ else
+ {
+ Assert.Equal(0, mode switch
+ {
+ ReadWriteMode.SyncArray => readable.Read(new byte[1], 0, 1),
+ ReadWriteMode.SyncSpan => readable.Read(new byte[1]),
+ ReadWriteMode.AsyncArray => await readable.ReadAsync(new byte[1], 0, 1),
+ ReadWriteMode.AsyncMemory => await readable.ReadAsync(new byte[1]),
+ ReadWriteMode.SyncAPM => readable.EndRead(readable.BeginRead(new byte[1], 0, 1, null, null)),
+ ReadWriteMode.AsyncAPM => await Task.Factory.FromAsync(readable.BeginRead, readable.EndRead, new byte[1], 0, 1, null),
+ _ => throw new Exception($"Unknown mode: {mode}"),
+ });
+ }
+ }
+
+ [Theory]
+ [InlineData(ReadWriteMode.SyncArray)]
+ [InlineData(ReadWriteMode.AsyncArray)]
+ [InlineData(ReadWriteMode.AsyncAPM)]
+ public virtual async Task Read_DataStoredAtDesiredOffset(ReadWriteMode mode)
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ (Stream writeable, Stream readable) = GetReadWritePair(streams);
+
+ byte[] buffer = new byte[10];
+ int offset = 2;
+ byte value = 42;
+
+ Task write = Task.Run(() =>
+ {
+ writeable.WriteByte(value);
+ writeable.Dispose();
+ });
+
+ Assert.Equal(1, mode switch
+ {
+ ReadWriteMode.SyncArray => readable.Read(buffer, offset, buffer.Length - offset),
+ ReadWriteMode.AsyncArray => await readable.ReadAsync(buffer, offset, buffer.Length - offset),
+ ReadWriteMode.AsyncAPM => await Task.Factory.FromAsync(readable.BeginRead, readable.EndRead, buffer, offset, buffer.Length - offset, null),
+ _ => throw new Exception($"Unknown mode: {mode}"),
+ });
+
+ await write;
+
+ for (int i = 0; i < buffer.Length; i++)
+ {
+ Assert.Equal(i == offset ? value : 0, buffer[i]);
+ }
+ }
+
+ [Theory]
+ [InlineData(ReadWriteMode.SyncArray)]
+ [InlineData(ReadWriteMode.AsyncArray)]
+ [InlineData(ReadWriteMode.AsyncAPM)]
+ public virtual async Task Write_DataReadFromDesiredOffset(ReadWriteMode mode)
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ (Stream writeable, Stream readable) = GetReadWritePair(streams);
+
+ byte[] buffer = new[] { (byte)'\0', (byte)'\0', (byte)'h', (byte)'e', (byte)'l', (byte)'l', (byte)'o', (byte)'\0', (byte)'\0' };
+ const int Offset = 2, Count = 5;
+
+ Task write = Task.Run(async () =>
+ {
+ switch (mode)
+ {
+ case ReadWriteMode.SyncArray:
+ writeable.Write(buffer, Offset, Count);
+ break;
+
+ case ReadWriteMode.AsyncArray:
+ await writeable.WriteAsync(buffer, Offset, Count);
+ break;
+
+ case ReadWriteMode.AsyncAPM:
+ await Task.Factory.FromAsync(writeable.BeginWrite, writeable.EndWrite, buffer, Offset, Count, null);
+ break;
+
+ default:
+ throw new Exception($"Unknown mode: {mode}");
+ }
+
+ writeable.Dispose();
+ });
+
+ Assert.Equal("hello", new StreamReader(readable).ReadToEnd());
+ }
+
+ [Fact]
+ public virtual async Task WriteWithBrokenPipe_Throws()
+ {
+ if (!BrokenPipePropagatedImmediately)
+ {
+ return;
+ }
+
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ (Stream writeable, Stream readable) = GetReadWritePair(streams);
+
+ readable.Dispose();
+ byte[] buffer = new byte[4];
+
+ Assert.Throws(() => writeable.WriteByte(123));
+ Assert.Throws(() => writeable.Write(buffer));
+ Assert.Throws(() => writeable.Write(buffer, 0, buffer.Length));
+ await Assert.ThrowsAsync(() => writeable.WriteAsync(buffer).AsTask());
+ await Assert.ThrowsAsync(() => writeable.WriteAsync(buffer, 0, buffer.Length));
+ Assert.Throws(() => writeable.EndWrite(writeable.BeginWrite(buffer, 0, buffer.Length, null, null)));
+ await Assert.ThrowsAsync(() => Task.Factory.FromAsync(writeable.BeginWrite, writeable.EndWrite, buffer, 0, buffer.Length, null));
+ Assert.Throws(() => writeable.Flush());
+ }
+
+ [Fact]
+ public virtual async Task ReadAsync_NonReusableValueTask_AwaitMultipleTimes_Throws()
+ {
+ if (!ReadWriteValueTasksProtectSingleConsumption)
+ {
+ return;
+ }
+
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ var bytes = new byte[1];
+
+ ValueTask r = readable.ReadAsync(bytes);
+ await writeable.WriteAsync(new byte[] { 42 });
+ if (FlushRequiredToWriteData)
+ {
+ await writeable.FlushAsync();
+ }
+ Assert.Equal(1, await r);
+ Assert.Equal(42, bytes[0]);
+
+ await Assert.ThrowsAsync(async () => await r);
+ Assert.Throws(() => r.GetAwaiter().IsCompleted);
+ Assert.Throws(() => r.GetAwaiter().OnCompleted(() => { }));
+ Assert.Throws(() => r.GetAwaiter().GetResult());
+ }
+ }
+
+ [Fact]
+ public virtual async Task ReadAsync_NonReusableValueTask_MultipleContinuations_Throws()
+ {
+ if (!ReadWriteValueTasksProtectSingleConsumption)
+ {
+ return;
+ }
+
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ var b = new byte[1];
+ ValueTask r = readable.ReadAsync(b);
+ r.GetAwaiter().OnCompleted(() => { });
+ Assert.Throws(() => r.GetAwaiter().OnCompleted(() => { }));
+ }
+ }
+
+ public static IEnumerable ReadAsync_ContinuesOnCurrentContextIfDesired_MemberData() =>
+ from flowExecutionContext in new[] { true, false }
+ from continueOnCapturedContext in new bool?[] { null, false, true }
+ select new object[] { flowExecutionContext, continueOnCapturedContext };
+
+ [Theory]
+ [MemberData(nameof(ReadAsync_ContinuesOnCurrentContextIfDesired_MemberData))]
+ public virtual async Task ReadAsync_ContinuesOnCurrentSynchronizationContextIfDesired(bool flowExecutionContext, bool? continueOnCapturedContext)
+ {
+ await default(JumpToThreadPoolAwaiter); // escape xunit sync ctx
+
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ Assert.Null(SynchronizationContext.Current);
+
+ var continuationRan = new TaskCompletionSource();
+ var asyncLocal = new AsyncLocal();
+ bool schedulerWasFlowed = false;
+ bool executionContextWasFlowed = false;
+ Action continuation = () =>
+ {
+ schedulerWasFlowed = SynchronizationContext.Current is CustomSynchronizationContext;
+ executionContextWasFlowed = 42 == asyncLocal.Value;
+ continuationRan.SetResult(true);
+ };
+
+ var readBuffer = new byte[1];
+ ValueTask readValueTask = readable.ReadAsync(new byte[1]);
+
+ SynchronizationContext.SetSynchronizationContext(new CustomSynchronizationContext());
+ asyncLocal.Value = 42;
+ switch (continueOnCapturedContext)
+ {
+ case null:
+ if (flowExecutionContext)
+ {
+ readValueTask.GetAwaiter().OnCompleted(continuation);
+ }
+ else
+ {
+ readValueTask.GetAwaiter().UnsafeOnCompleted(continuation);
+ }
+ break;
+ default:
+ if (flowExecutionContext)
+ {
+ readValueTask.ConfigureAwait(continueOnCapturedContext.Value).GetAwaiter().OnCompleted(continuation);
+ }
+ else
+ {
+ readValueTask.ConfigureAwait(continueOnCapturedContext.Value).GetAwaiter().UnsafeOnCompleted(continuation);
+ }
+ break;
+ }
+ asyncLocal.Value = 0;
+ SynchronizationContext.SetSynchronizationContext(null);
+
+ Assert.False(readValueTask.IsCompleted);
+ Assert.False(readValueTask.IsCompletedSuccessfully);
+ await writeable.WriteAsync(new byte[] { 42 });
+ if (FlushRequiredToWriteData)
+ {
+ if (FlushGuaranteesAllDataWritten)
+ {
+ await writeable.FlushAsync();
+ }
+ else
+ {
+ await writeable.DisposeAsync();
+ }
+ }
+
+ await continuationRan.Task;
+ Assert.True(readValueTask.IsCompleted);
+ Assert.True(readValueTask.IsCompletedSuccessfully);
+
+ Assert.Equal(continueOnCapturedContext != false, schedulerWasFlowed);
+ Assert.Equal(flowExecutionContext, executionContextWasFlowed);
+
+ if (!FlushGuaranteesAllDataWritten)
+ {
+ break;
+ }
+ }
+ }
+
+ [Theory]
+ [MemberData(nameof(ReadAsync_ContinuesOnCurrentContextIfDesired_MemberData))]
+ public virtual async Task ReadAsync_ContinuesOnCurrentTaskSchedulerIfDesired(bool flowExecutionContext, bool? continueOnCapturedContext)
+ {
+ await default(JumpToThreadPoolAwaiter); // escape xunit sync ctx
+
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ Assert.Null(SynchronizationContext.Current);
+
+ var continuationRan = new TaskCompletionSource();
+ var asyncLocal = new AsyncLocal();
+ bool schedulerWasFlowed = false;
+ bool executionContextWasFlowed = false;
+ Action continuation = () =>
+ {
+ schedulerWasFlowed = TaskScheduler.Current is CustomTaskScheduler;
+ executionContextWasFlowed = 42 == asyncLocal.Value;
+ continuationRan.SetResult();
+ };
+
+ var readBuffer = new byte[1];
+ ValueTask readValueTask = readable.ReadAsync(new byte[1]);
+
+ await Task.Factory.StartNew(() =>
+ {
+ Assert.IsType(TaskScheduler.Current);
+ asyncLocal.Value = 42;
+ switch (continueOnCapturedContext)
+ {
+ case null:
+ if (flowExecutionContext)
+ {
+ readValueTask.GetAwaiter().OnCompleted(continuation);
+ }
+ else
+ {
+ readValueTask.GetAwaiter().UnsafeOnCompleted(continuation);
+ }
+ break;
+ default:
+ if (flowExecutionContext)
+ {
+ readValueTask.ConfigureAwait(continueOnCapturedContext.Value).GetAwaiter().OnCompleted(continuation);
+ }
+ else
+ {
+ readValueTask.ConfigureAwait(continueOnCapturedContext.Value).GetAwaiter().UnsafeOnCompleted(continuation);
+ }
+ break;
+ }
+ asyncLocal.Value = 0;
+ }, CancellationToken.None, TaskCreationOptions.None, new CustomTaskScheduler());
+
+ Assert.False(readValueTask.IsCompleted);
+ Assert.False(readValueTask.IsCompletedSuccessfully);
+ await writeable.WriteAsync(new byte[] { 42 });
+ if (FlushRequiredToWriteData)
+ {
+ if (FlushGuaranteesAllDataWritten)
+ {
+ await writeable.FlushAsync();
+ }
+ else
+ {
+ await writeable.DisposeAsync();
+ }
+ }
+
+ await continuationRan.Task;
+ Assert.True(readValueTask.IsCompleted);
+ Assert.True(readValueTask.IsCompletedSuccessfully);
+
+ Assert.Equal(continueOnCapturedContext != false, schedulerWasFlowed);
+ Assert.Equal(flowExecutionContext, executionContextWasFlowed);
+
+ if (!FlushGuaranteesAllDataWritten)
+ {
+ break;
+ }
+ }
+ }
+
+ [Theory]
+ [InlineData(ReadWriteMode.SyncArray)]
+ [InlineData(ReadWriteMode.SyncSpan)]
+ [InlineData(ReadWriteMode.AsyncArray)]
+ [InlineData(ReadWriteMode.AsyncMemory)]
+ [InlineData(ReadWriteMode.SyncAPM)]
+ [InlineData(ReadWriteMode.AsyncAPM)]
+ public virtual async Task ZeroByteRead_BlocksUntilDataAvailableOrNops(ReadWriteMode mode)
+ {
+ using StreamPair streams = await CreateConnectedStreamsAsync();
+ foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+ {
+ for (int iter = 0; iter < 2; iter++)
+ {
+ Task zeroByteRead = mode switch
+ {
+ ReadWriteMode.SyncSpan => Task.Run(() => readable.Read(Span.Empty)),
+ ReadWriteMode.SyncArray => Task.Run(() => readable.Read(new byte[0], 0, 0)),
+ ReadWriteMode.AsyncArray => readable.ReadAsync(new byte[0], 0, 0),
+ ReadWriteMode.AsyncMemory => readable.ReadAsync(Memory.Empty).AsTask(),
+ ReadWriteMode.SyncAPM => Task.Run(() => readable.EndRead(readable.BeginRead(Array.Empty(), 0, 0, null, null))),
+ ReadWriteMode.AsyncAPM => Task.Factory.FromAsync(readable.BeginRead, readable.EndRead, Array.Empty