diff --git a/src/Http/Http.Abstractions/ref/Microsoft.AspNetCore.Http.Abstractions.netcoreapp3.0.cs b/src/Http/Http.Abstractions/ref/Microsoft.AspNetCore.Http.Abstractions.netcoreapp3.0.cs
index 9dd9e01997..819234838c 100644
--- a/src/Http/Http.Abstractions/ref/Microsoft.AspNetCore.Http.Abstractions.netcoreapp3.0.cs
+++ b/src/Http/Http.Abstractions/ref/Microsoft.AspNetCore.Http.Abstractions.netcoreapp3.0.cs
@@ -250,7 +250,7 @@ namespace Microsoft.AspNetCore.Http
{
protected HttpRequest() { }
public abstract System.IO.Stream Body { get; set; }
- public virtual System.IO.Pipelines.PipeReader BodyReader { get { throw null; } set { } }
+ public virtual System.IO.Pipelines.PipeReader BodyReader { get { throw null; } }
public abstract long? ContentLength { get; set; }
public abstract string ContentType { get; set; }
public abstract Microsoft.AspNetCore.Http.IRequestCookieCollection Cookies { get; set; }
@@ -274,7 +274,7 @@ namespace Microsoft.AspNetCore.Http
{
protected HttpResponse() { }
public abstract System.IO.Stream Body { get; set; }
- public virtual System.IO.Pipelines.PipeWriter BodyWriter { get { throw null; } set { } }
+ public virtual System.IO.Pipelines.PipeWriter BodyWriter { get { throw null; } }
public abstract long? ContentLength { get; set; }
public abstract string ContentType { get; set; }
public abstract Microsoft.AspNetCore.Http.IResponseCookies Cookies { get; }
diff --git a/src/Http/Http.Abstractions/src/HttpRequest.cs b/src/Http/Http.Abstractions/src/HttpRequest.cs
index a63438450b..4a6c58b7e0 100644
--- a/src/Http/Http.Abstractions/src/HttpRequest.cs
+++ b/src/Http/Http.Abstractions/src/HttpRequest.cs
@@ -105,9 +105,9 @@ namespace Microsoft.AspNetCore.Http
public abstract Stream Body { get; set; }
///
- /// Gets or sets the request body pipe .
+ /// Gets the request body pipe .
///
- public virtual PipeReader BodyReader { get => throw new NotImplementedException(); set => throw new NotImplementedException(); }
+ public virtual PipeReader BodyReader { get => throw new NotImplementedException(); }
///
/// Checks the Content-Type header for form types.
diff --git a/src/Http/Http.Abstractions/src/HttpResponse.cs b/src/Http/Http.Abstractions/src/HttpResponse.cs
index d5a144dd7d..9df4f095e6 100644
--- a/src/Http/Http.Abstractions/src/HttpResponse.cs
+++ b/src/Http/Http.Abstractions/src/HttpResponse.cs
@@ -45,9 +45,9 @@ namespace Microsoft.AspNetCore.Http
public abstract Stream Body { get; set; }
///
- /// Gets or sets the response body pipe
+ /// Gets the response body pipe
///
- public virtual PipeWriter BodyWriter { get => throw new NotImplementedException(); set => throw new NotImplementedException(); }
+ public virtual PipeWriter BodyWriter { get => throw new NotImplementedException(); }
///
/// Gets or sets the value for the Content-Length response header.
diff --git a/src/Http/Http.Abstractions/test/HttpResponseWritingExtensionsTests.cs b/src/Http/Http.Abstractions/test/HttpResponseWritingExtensionsTests.cs
index 70684be914..0591abdce6 100644
--- a/src/Http/Http.Abstractions/test/HttpResponseWritingExtensionsTests.cs
+++ b/src/Http/Http.Abstractions/test/HttpResponseWritingExtensionsTests.cs
@@ -3,8 +3,7 @@
using System;
using System.IO;
-using System.IO.Pipelines;
-using System.IO.Pipelines.Tests;
+using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Xunit;
@@ -36,15 +35,12 @@ namespace Microsoft.AspNetCore.Http
[MemberData(nameof(Encodings))]
public async Task WritingTextThatRequiresMultipleSegmentsWorks(Encoding encoding)
{
- // Need to change the StreamPipeWriter with a capped MemoryPool
- var memoryPool = new TestMemoryPool(maxBufferSize: 16);
var outputStream = new MemoryStream();
- var streamPipeWriter = new StreamPipeWriter(outputStream, minimumSegmentSize: 0, memoryPool);
HttpContext context = new DefaultHttpContext();
- context.Response.BodyWriter = streamPipeWriter;
+ context.Response.Body = outputStream;
- var inputString = "昨日すき焼きを食べました";
+ var inputString = string.Concat(Enumerable.Repeat("昨日すき焼きを食べました", 1000));
var expected = encoding.GetBytes(inputString);
await context.Response.WriteAsync(inputString, encoding);
@@ -52,11 +48,8 @@ namespace Microsoft.AspNetCore.Http
var actual = new byte[expected.Length];
var length = outputStream.Read(actual);
- var res1 = encoding.GetString(actual);
- var res2 = encoding.GetString(expected);
Assert.Equal(expected.Length, length);
Assert.Equal(expected, actual);
- streamPipeWriter.Complete();
}
[Theory]
diff --git a/src/Http/Http.Features/ref/Microsoft.AspNetCore.Http.Features.netstandard2.0.cs b/src/Http/Http.Features/ref/Microsoft.AspNetCore.Http.Features.netstandard2.0.cs
index 8c0f10b5dc..c97e11f219 100644
--- a/src/Http/Http.Features/ref/Microsoft.AspNetCore.Http.Features.netstandard2.0.cs
+++ b/src/Http/Http.Features/ref/Microsoft.AspNetCore.Http.Features.netstandard2.0.cs
@@ -246,7 +246,7 @@ namespace Microsoft.AspNetCore.Http.Features
}
public partial interface IRequestBodyPipeFeature
{
- System.IO.Pipelines.PipeReader Reader { get; set; }
+ System.IO.Pipelines.PipeReader Reader { get; }
}
public partial interface IRequestCookiesFeature
{
@@ -254,7 +254,7 @@ namespace Microsoft.AspNetCore.Http.Features
}
public partial interface IResponseBodyPipeFeature
{
- System.IO.Pipelines.PipeWriter Writer { get; set; }
+ System.IO.Pipelines.PipeWriter Writer { get; }
}
public partial interface IResponseCookiesFeature
{
diff --git a/src/Http/Http.Features/src/IRequestBodyPipeFeature.cs b/src/Http/Http.Features/src/IRequestBodyPipeFeature.cs
index 9357edb421..99e109b99d 100644
--- a/src/Http/Http.Features/src/IRequestBodyPipeFeature.cs
+++ b/src/Http/Http.Features/src/IRequestBodyPipeFeature.cs
@@ -13,6 +13,6 @@ namespace Microsoft.AspNetCore.Http.Features
///
/// A representing the request body, if any.
///
- PipeReader Reader { get; set; }
+ PipeReader Reader { get; }
}
}
diff --git a/src/Http/Http.Features/src/IResponseBodyPipeFeature.cs b/src/Http/Http.Features/src/IResponseBodyPipeFeature.cs
index 71a55d12b9..45d0e35f0a 100644
--- a/src/Http/Http.Features/src/IResponseBodyPipeFeature.cs
+++ b/src/Http/Http.Features/src/IResponseBodyPipeFeature.cs
@@ -14,6 +14,6 @@ namespace Microsoft.AspNetCore.Http.Features
///
/// A representing the response body, if any.
///
- PipeWriter Writer { get; set; }
+ PipeWriter Writer { get; }
}
}
diff --git a/src/Http/Http/perf/NoopStream.cs b/src/Http/Http/perf/NoopStream.cs
deleted file mode 100644
index 9dcfe8408f..0000000000
--- a/src/Http/Http/perf/NoopStream.cs
+++ /dev/null
@@ -1,70 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System;
-using System.IO;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace Microsoft.AspNetCore.Http
-{
- public class NoopStream : Stream
- {
- public override bool CanRead => true;
-
- public override bool CanSeek => throw new NotImplementedException();
-
- public override bool CanWrite => true;
-
- public override long Length => throw new NotImplementedException();
-
- public override long Position { get => throw new NotImplementedException(); set => throw new NotImplementedException(); }
-
- public override void Flush()
- {
- }
-
- public override int Read(byte[] buffer, int offset, int count)
- {
- return 0;
- }
-
- public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- return Task.FromResult(0);
- }
-
- public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
- {
- return new ValueTask(0);
- }
-
- public override long Seek(long offset, SeekOrigin origin)
- {
- throw new NotImplementedException();
- }
-
- public override void SetLength(long value)
- {
- }
-
- public override void Write(byte[] buffer, int offset, int count)
- {
- }
-
- public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- return Task.CompletedTask;
- }
-
- public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default(CancellationToken))
- {
- return default(ValueTask);
- }
-
- public override Task FlushAsync(CancellationToken cancellationToken)
- {
- return Task.CompletedTask;
- }
- }
-}
diff --git a/src/Http/Http/perf/StreamPipeReaderBenchmark.cs b/src/Http/Http/perf/StreamPipeReaderBenchmark.cs
deleted file mode 100644
index 2b5ef7c24b..0000000000
--- a/src/Http/Http/perf/StreamPipeReaderBenchmark.cs
+++ /dev/null
@@ -1,82 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System;
-using System.IO.Pipelines;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using BenchmarkDotNet.Attributes;
-
-namespace Microsoft.AspNetCore.Http
-{
- public class StreamPipeReaderBenchmark
- {
- private StreamPipeReader _pipeReaderNoop;
- private StreamPipeReader _pipeReaderHelloWorld;
- private StreamPipeReader _pipeReaderHelloWorldAync;
-
- [IterationSetup]
- public void Setup()
- {
- _pipeReaderNoop = new StreamPipeReader(new NoopStream());
- _pipeReaderHelloWorld = new StreamPipeReader(new HelloWorldStream());
- _pipeReaderHelloWorldAync = new StreamPipeReader(new HelloWorldAsyncStream());
- }
-
- [Benchmark]
- public async Task ReadNoop()
- {
- await _pipeReaderNoop.ReadAsync();
- }
-
- [Benchmark]
- public async Task ReadHelloWorld()
- {
- var result = await _pipeReaderHelloWorld.ReadAsync();
- _pipeReaderHelloWorld.AdvanceTo(result.Buffer.End);
- }
-
- [Benchmark]
- public async Task ReadHelloWorldAsync()
- {
- var result = await _pipeReaderHelloWorldAync.ReadAsync();
- _pipeReaderHelloWorldAync.AdvanceTo(result.Buffer.End);
- }
-
- private class HelloWorldStream : NoopStream
- {
- private static byte[] bytes = Encoding.ASCII.GetBytes("Hello World");
- public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- bytes.CopyTo(buffer, 0);
- return Task.FromResult(11);
- }
-
- public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
- {
- bytes.CopyTo(buffer);
-
- return new ValueTask(11);
- }
- }
-
- private class HelloWorldAsyncStream : NoopStream
- {
- private static byte[] bytes = Encoding.ASCII.GetBytes("Hello World");
- public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- await Task.Yield();
- bytes.CopyTo(buffer, 0);
- return 11;
- }
-
- public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
- {
- await Task.Yield();
- bytes.CopyTo(buffer);
- return 11;
- }
- }
- }
-}
diff --git a/src/Http/Http/perf/StreamPipeWriterBenchmark.cs b/src/Http/Http/perf/StreamPipeWriterBenchmark.cs
deleted file mode 100644
index dd7836120d..0000000000
--- a/src/Http/Http/perf/StreamPipeWriterBenchmark.cs
+++ /dev/null
@@ -1,38 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System.IO;
-using System.IO.Pipelines;
-using System.Text;
-using System.Threading.Tasks;
-using BenchmarkDotNet.Attributes;
-
-namespace Microsoft.AspNetCore.Http
-{
- public class StreamPipeWriterBenchmark
- {
- private Stream _memoryStream;
- private StreamPipeWriter _pipeWriter;
- private static byte[] _helloWorldBytes = Encoding.ASCII.GetBytes("Hello World");
- private static byte[] _largeWrite = Encoding.ASCII.GetBytes(new string('a', 50000));
-
- [IterationSetup]
- public void Setup()
- {
- _memoryStream = new NoopStream();
- _pipeWriter = new StreamPipeWriter(_memoryStream);
- }
-
- [Benchmark]
- public async Task WriteHelloWorld()
- {
- await _pipeWriter.WriteAsync(_helloWorldBytes);
- }
-
- [Benchmark]
- public async Task WriteHelloWorldLargeWrite()
- {
- await _pipeWriter.WriteAsync(_largeWrite);
- }
- }
-}
diff --git a/src/Http/Http/ref/Microsoft.AspNetCore.Http.netcoreapp3.0.cs b/src/Http/Http/ref/Microsoft.AspNetCore.Http.netcoreapp3.0.cs
index 16ad1810c3..2c299f83d9 100644
--- a/src/Http/Http/ref/Microsoft.AspNetCore.Http.netcoreapp3.0.cs
+++ b/src/Http/Http/ref/Microsoft.AspNetCore.Http.netcoreapp3.0.cs
@@ -236,7 +236,7 @@ namespace Microsoft.AspNetCore.Http.Features
public partial class RequestBodyPipeFeature : Microsoft.AspNetCore.Http.Features.IRequestBodyPipeFeature
{
public RequestBodyPipeFeature(Microsoft.AspNetCore.Http.HttpContext context) { }
- public System.IO.Pipelines.PipeReader Reader { get { throw null; } set { } }
+ public System.IO.Pipelines.PipeReader Reader { get { throw null; } }
}
public partial class RequestCookiesFeature : Microsoft.AspNetCore.Http.Features.IRequestCookiesFeature
{
@@ -254,7 +254,7 @@ namespace Microsoft.AspNetCore.Http.Features
public partial class ResponseBodyPipeFeature : Microsoft.AspNetCore.Http.Features.IResponseBodyPipeFeature
{
public ResponseBodyPipeFeature(Microsoft.AspNetCore.Http.HttpContext context) { }
- public System.IO.Pipelines.PipeWriter Writer { get { throw null; } set { } }
+ public System.IO.Pipelines.PipeWriter Writer { get { throw null; } }
}
public partial class ResponseCookiesFeature : Microsoft.AspNetCore.Http.Features.IResponseCookiesFeature
{
@@ -326,7 +326,7 @@ namespace Microsoft.AspNetCore.Http.Internal
{
public DefaultHttpRequest(Microsoft.AspNetCore.Http.DefaultHttpContext context) { }
public override System.IO.Stream Body { get { throw null; } set { } }
- public override System.IO.Pipelines.PipeReader BodyReader { get { throw null; } set { } }
+ public override System.IO.Pipelines.PipeReader BodyReader { get { throw null; } }
public override long? ContentLength { get { throw null; } set { } }
public override string ContentType { get { throw null; } set { } }
public override Microsoft.AspNetCore.Http.IRequestCookieCollection Cookies { get { throw null; } set { } }
@@ -353,7 +353,7 @@ namespace Microsoft.AspNetCore.Http.Internal
{
public DefaultHttpResponse(Microsoft.AspNetCore.Http.DefaultHttpContext context) { }
public override System.IO.Stream Body { get { throw null; } set { } }
- public override System.IO.Pipelines.PipeWriter BodyWriter { get { throw null; } set { } }
+ public override System.IO.Pipelines.PipeWriter BodyWriter { get { throw null; } }
public override long? ContentLength { get { throw null; } set { } }
public override string ContentType { get { throw null; } set { } }
public override Microsoft.AspNetCore.Http.IResponseCookies Cookies { get { throw null; } }
@@ -492,93 +492,3 @@ namespace Microsoft.Extensions.DependencyInjection
public static Microsoft.Extensions.DependencyInjection.IServiceCollection AddHttpContextAccessor(this Microsoft.Extensions.DependencyInjection.IServiceCollection services) { throw null; }
}
}
-namespace System.IO.Pipelines
-{
- public partial class ReadOnlyPipeStream : System.IO.Stream
- {
- public ReadOnlyPipeStream(System.IO.Pipelines.PipeReader pipeReader) { }
- public ReadOnlyPipeStream(System.IO.Pipelines.PipeReader pipeReader, bool allowSynchronousIO) { }
- public override bool CanRead { get { throw null; } }
- public override bool CanSeek { get { throw null; } }
- public override bool CanWrite { get { throw null; } }
- public System.IO.Pipelines.PipeReader InnerPipeReader { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } }
- public override long Length { get { throw null; } }
- public override long Position { get { throw null; } set { } }
- public override int WriteTimeout { get { throw null; } set { } }
- public override System.IAsyncResult BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) { throw null; }
- public override System.Threading.Tasks.Task CopyToAsync(System.IO.Stream destination, int bufferSize, System.Threading.CancellationToken cancellationToken) { throw null; }
- public override int EndRead(System.IAsyncResult asyncResult) { throw null; }
- public override void Flush() { }
- public override System.Threading.Tasks.Task FlushAsync(System.Threading.CancellationToken cancellationToken) { throw null; }
- public override int Read(byte[] buffer, int offset, int count) { throw null; }
- public override System.Threading.Tasks.Task ReadAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
- public override System.Threading.Tasks.ValueTask ReadAsync(System.Memory destination, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
- public override long Seek(long offset, System.IO.SeekOrigin origin) { throw null; }
- public override void SetLength(long value) { }
- public override void Write(byte[] buffer, int offset, int count) { }
- public override System.Threading.Tasks.Task WriteAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
- }
- public partial class StreamPipeReader : System.IO.Pipelines.PipeReader, System.IDisposable
- {
- public StreamPipeReader(System.IO.Stream readingStream) { }
- public StreamPipeReader(System.IO.Stream readingStream, System.IO.Pipelines.StreamPipeReaderAdapterOptions options) { }
- public System.IO.Stream InnerStream { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } }
- public override void AdvanceTo(System.SequencePosition consumed) { }
- public override void AdvanceTo(System.SequencePosition consumed, System.SequencePosition examined) { }
- public override void CancelPendingRead() { }
- public override void Complete(System.Exception exception = null) { }
- public void Dispose() { }
- public override void OnWriterCompleted(System.Action callback, object state) { }
- [System.Diagnostics.DebuggerStepThroughAttribute]
- public override System.Threading.Tasks.ValueTask ReadAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
- public override bool TryRead(out System.IO.Pipelines.ReadResult result) { throw null; }
- }
- public partial class StreamPipeReaderAdapterOptions
- {
- public const int DefaultMinimumReadThreshold = 256;
- public const int DefaultMinimumSegmentSize = 4096;
- public static System.IO.Pipelines.StreamPipeReaderAdapterOptions DefaultOptions;
- public StreamPipeReaderAdapterOptions() { }
- public StreamPipeReaderAdapterOptions(int minimumSegmentSize, int minimumReadThreshold, System.Buffers.MemoryPool memoryPool) { }
- public System.Buffers.MemoryPool MemoryPool { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } [System.Runtime.CompilerServices.CompilerGeneratedAttribute]set { } }
- public int MinimumReadThreshold { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } [System.Runtime.CompilerServices.CompilerGeneratedAttribute]set { } }
- public int MinimumSegmentSize { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } [System.Runtime.CompilerServices.CompilerGeneratedAttribute]set { } }
- }
- public partial class StreamPipeWriter : System.IO.Pipelines.PipeWriter, System.IDisposable
- {
- public StreamPipeWriter(System.IO.Stream writingStream) { }
- public StreamPipeWriter(System.IO.Stream writingStream, int minimumSegmentSize, System.Buffers.MemoryPool pool = null) { }
- public System.IO.Stream InnerStream { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } }
- public override void Advance(int count) { }
- public override void CancelPendingFlush() { }
- public override void Complete(System.Exception exception = null) { }
- public void Dispose() { }
- public override System.Threading.Tasks.ValueTask FlushAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
- public override System.Memory GetMemory(int sizeHint = 0) { throw null; }
- public override System.Span GetSpan(int sizeHint = 0) { throw null; }
- public override void OnReaderCompleted(System.Action callback, object state) { }
- }
- public partial class WriteOnlyPipeStream : System.IO.Stream
- {
- public WriteOnlyPipeStream(System.IO.Pipelines.PipeWriter pipeWriter) { }
- public WriteOnlyPipeStream(System.IO.Pipelines.PipeWriter pipeWriter, bool allowSynchronousIO) { }
- public override bool CanRead { get { throw null; } }
- public override bool CanSeek { get { throw null; } }
- public override bool CanWrite { get { throw null; } }
- public System.IO.Pipelines.PipeWriter InnerPipeWriter { [System.Runtime.CompilerServices.CompilerGeneratedAttribute]get { throw null; } }
- public override long Length { get { throw null; } }
- public override long Position { get { throw null; } set { } }
- public override int ReadTimeout { get { throw null; } set { } }
- public override System.IAsyncResult BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) { throw null; }
- public override void EndWrite(System.IAsyncResult asyncResult) { }
- public override void Flush() { }
- public override System.Threading.Tasks.Task FlushAsync(System.Threading.CancellationToken cancellationToken) { throw null; }
- public override int Read(byte[] buffer, int offset, int count) { throw null; }
- public override System.Threading.Tasks.Task ReadAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
- public override long Seek(long offset, System.IO.SeekOrigin origin) { throw null; }
- public override void SetLength(long value) { }
- public override void Write(byte[] buffer, int offset, int count) { }
- public override System.Threading.Tasks.Task WriteAsync(byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) { throw null; }
- public override System.Threading.Tasks.ValueTask WriteAsync(System.ReadOnlyMemory source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
- }
-}
diff --git a/src/Http/Http/src/Features/RequestBodyPipeFeature.cs b/src/Http/Http/src/Features/RequestBodyPipeFeature.cs
index ee86031fde..61154c5f0a 100644
--- a/src/Http/Http/src/Features/RequestBodyPipeFeature.cs
+++ b/src/Http/Http/src/Features/RequestBodyPipeFeature.cs
@@ -2,14 +2,16 @@
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
+using System.IO;
using System.IO.Pipelines;
+using System.Threading.Tasks;
namespace Microsoft.AspNetCore.Http.Features
{
public class RequestBodyPipeFeature : IRequestBodyPipeFeature
{
- private StreamPipeReader _internalPipeReader;
- private PipeReader _userSetPipeReader;
+ private PipeReader _internalPipeReader;
+ private Stream _streamInstanceWhenWrapped;
private HttpContext _context;
public RequestBodyPipeFeature(HttpContext context)
@@ -25,25 +27,21 @@ namespace Microsoft.AspNetCore.Http.Features
{
get
{
- if (_userSetPipeReader != null)
- {
- return _userSetPipeReader;
- }
-
if (_internalPipeReader == null ||
- !object.ReferenceEquals(_internalPipeReader.InnerStream, _context.Request.Body))
+ !ReferenceEquals(_streamInstanceWhenWrapped, _context.Request.Body))
{
- _internalPipeReader = new StreamPipeReader(_context.Request.Body);
- _context.Response.RegisterForDispose(_internalPipeReader);
+ _streamInstanceWhenWrapped = _context.Request.Body;
+ _internalPipeReader = PipeReader.Create(_context.Request.Body);
+
+ _context.Response.OnCompleted((self) =>
+ {
+ ((PipeReader)self).Complete();
+ return Task.CompletedTask;
+ }, _internalPipeReader);
}
return _internalPipeReader;
}
- set
- {
- _userSetPipeReader = value ?? throw new ArgumentNullException(nameof(value));
- // TODO set the request body Stream to an adapted pipe https://github.com/aspnet/AspNetCore/issues/3971
- }
}
}
}
diff --git a/src/Http/Http/src/Features/ResponseBodyPipeFeature.cs b/src/Http/Http/src/Features/ResponseBodyPipeFeature.cs
index 0699679365..2de8c86b9c 100644
--- a/src/Http/Http/src/Features/ResponseBodyPipeFeature.cs
+++ b/src/Http/Http/src/Features/ResponseBodyPipeFeature.cs
@@ -2,14 +2,16 @@
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
+using System.IO;
using System.IO.Pipelines;
+using System.Threading.Tasks;
namespace Microsoft.AspNetCore.Http.Features
{
public class ResponseBodyPipeFeature : IResponseBodyPipeFeature
{
- private StreamPipeWriter _internalPipeWriter;
- private PipeWriter _userSetPipeWriter;
+ private PipeWriter _internalPipeWriter;
+ private Stream _streamInstanceWhenWrapped;
private HttpContext _context;
public ResponseBodyPipeFeature(HttpContext context)
@@ -25,25 +27,21 @@ namespace Microsoft.AspNetCore.Http.Features
{
get
{
- if (_userSetPipeWriter != null)
- {
- return _userSetPipeWriter;
- }
-
if (_internalPipeWriter == null ||
- !object.ReferenceEquals(_internalPipeWriter.InnerStream, _context.Response.Body))
+ !ReferenceEquals(_streamInstanceWhenWrapped, _context.Response.Body))
{
- _internalPipeWriter = new StreamPipeWriter(_context.Response.Body);
- _context.Response.RegisterForDispose(_internalPipeWriter);
+ _streamInstanceWhenWrapped = _context.Response.Body;
+ _internalPipeWriter = PipeWriter.Create(_context.Response.Body);
+
+ _context.Response.OnCompleted((self) =>
+ {
+ ((PipeWriter)self).Complete();
+ return Task.CompletedTask;
+ }, _internalPipeWriter);
}
return _internalPipeWriter;
}
- set
- {
- _userSetPipeWriter = value ?? throw new ArgumentNullException(nameof(value));
- // TODO set the response body Stream to an adapted pipe https://github.com/aspnet/AspNetCore/issues/3971
- }
}
}
}
diff --git a/src/Http/Http/src/Internal/DefaultHttpRequest.cs b/src/Http/Http/src/Internal/DefaultHttpRequest.cs
index 964a478510..bc0a4c42dc 100644
--- a/src/Http/Http/src/Internal/DefaultHttpRequest.cs
+++ b/src/Http/Http/src/Internal/DefaultHttpRequest.cs
@@ -174,7 +174,6 @@ namespace Microsoft.AspNetCore.Http.Internal
public override PipeReader BodyReader
{
get { return RequestBodyPipeFeature.Reader; }
- set { RequestBodyPipeFeature.Reader = value; }
}
struct FeatureInterfaces
diff --git a/src/Http/Http/src/Internal/DefaultHttpResponse.cs b/src/Http/Http/src/Internal/DefaultHttpResponse.cs
index 068c4ec9bd..19e75c3aa0 100644
--- a/src/Http/Http/src/Internal/DefaultHttpResponse.cs
+++ b/src/Http/Http/src/Internal/DefaultHttpResponse.cs
@@ -112,7 +112,6 @@ namespace Microsoft.AspNetCore.Http.Internal
public override PipeWriter BodyWriter
{
get { return ResponseBodyPipeFeature.Writer; }
- set { ResponseBodyPipeFeature.Writer = value; }
}
public override void OnStarting(Func callback, object state)
diff --git a/src/Http/Http/src/ReadOnlyPipeStream.cs b/src/Http/Http/src/ReadOnlyPipeStream.cs
deleted file mode 100644
index 919745bc5d..0000000000
--- a/src/Http/Http/src/ReadOnlyPipeStream.cs
+++ /dev/null
@@ -1,247 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System;
-using System.Buffers;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace System.IO.Pipelines
-{
- ///
- /// Represents a read-only Stream backed by a PipeReader
- ///
- public class ReadOnlyPipeStream : Stream
- {
- private bool _allowSynchronousIO = true;
-
- ///
- /// Creates a new ReadOnlyPipeStream
- ///
- /// The PipeReader to read from.
- public ReadOnlyPipeStream(PipeReader pipeReader) :
- this(pipeReader, allowSynchronousIO: true)
- {
- }
-
- ///
- /// Creates a new ReadOnlyPipeStream
- ///
- /// The PipeReader to read from.
- /// Whether synchronous IO is allowed.
- public ReadOnlyPipeStream(PipeReader pipeReader, bool allowSynchronousIO)
- {
- _allowSynchronousIO = allowSynchronousIO;
- InnerPipeReader = pipeReader;
- }
-
- ///
- public override bool CanSeek => false;
-
- ///
- public override bool CanRead => true;
-
- ///
- public override bool CanWrite => false;
-
- ///
- public override long Length => throw new NotSupportedException();
-
- ///
- public override long Position
- {
- get => throw new NotSupportedException();
- set => throw new NotSupportedException();
- }
-
- ///
- public override int WriteTimeout
- {
- get => throw new NotSupportedException();
- set => throw new NotSupportedException();
- }
-
- public PipeReader InnerPipeReader { get; }
-
- ///
- public override void Write(byte[] buffer, int offset, int count)
- => throw new NotSupportedException();
-
- ///
- public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- => throw new NotSupportedException();
-
- ///
- public override void Flush()
- {
- throw new NotSupportedException();
- }
-
- ///
- public override Task FlushAsync(CancellationToken cancellationToken)
- {
- throw new NotSupportedException();
- }
-
- ///
- public override long Seek(long offset, SeekOrigin origin)
- {
- throw new NotSupportedException();
- }
-
- ///
- public override void SetLength(long value)
- {
- throw new NotSupportedException();
- }
-
- ///
- public override int Read(byte[] buffer, int offset, int count)
- {
- if (!_allowSynchronousIO)
- {
- ThrowHelper.ThrowInvalidOperationException_SynchronousReadsDisallowed();
- }
- return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
- }
-
- ///
- public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
- {
- var task = ReadAsync(buffer, offset, count, default, state);
- if (callback != null)
- {
- task.ContinueWith(t => callback.Invoke(t));
- }
- return task;
- }
-
- ///
- public override int EndRead(IAsyncResult asyncResult)
- {
- return ((Task)asyncResult).GetAwaiter().GetResult();
- }
-
- private Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object state)
- {
- var tcs = new TaskCompletionSource(state);
- var task = ReadAsync(buffer, offset, count, cancellationToken);
- task.ContinueWith((task2, state2) =>
- {
- var tcs2 = (TaskCompletionSource)state2;
- if (task2.IsCanceled)
- {
- tcs2.SetCanceled();
- }
- else if (task2.IsFaulted)
- {
- tcs2.SetException(task2.Exception);
- }
- else
- {
- tcs2.SetResult(task2.Result);
- }
- }, tcs, cancellationToken);
- return tcs.Task;
- }
-
- ///
- public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- return ReadAsyncInternal(new Memory(buffer, offset, count), cancellationToken).AsTask();
- }
-
- ///
- public override ValueTask ReadAsync(Memory destination, CancellationToken cancellationToken = default)
- {
- return ReadAsyncInternal(destination, cancellationToken);
- }
-
- private async ValueTask ReadAsyncInternal(Memory buffer, CancellationToken cancellationToken)
- {
- while (true)
- {
- var result = await InnerPipeReader.ReadAsync(cancellationToken);
-
- if (result.IsCanceled)
- {
- ThrowHelper.ThrowOperationCanceledException_ReadCanceled();
- }
-
- var readableBuffer = result.Buffer;
- var readableBufferLength = readableBuffer.Length;
-
- var consumed = readableBuffer.End;
- var actual = 0;
- try
- {
- if (readableBufferLength != 0)
- {
- actual = (int)Math.Min(readableBufferLength, buffer.Length);
-
- var slice = actual == readableBufferLength ? readableBuffer : readableBuffer.Slice(0, actual);
- consumed = slice.End;
- slice.CopyTo(buffer.Span);
-
- return actual;
- }
-
- if (result.IsCompleted)
- {
- return 0;
- }
- }
- finally
- {
- InnerPipeReader.AdvanceTo(consumed);
- }
- }
- }
-
- ///
- public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
- {
- if (destination == null)
- {
- throw new ArgumentNullException(nameof(destination));
- }
-
- if (bufferSize <= 0)
- {
- throw new ArgumentOutOfRangeException(nameof(bufferSize));
- }
-
- return CopyToAsyncInternal(destination, cancellationToken);
- }
-
- private async Task CopyToAsyncInternal(Stream destination, CancellationToken cancellationToken)
- {
- while (true)
- {
- var result = await InnerPipeReader.ReadAsync(cancellationToken);
- var readableBuffer = result.Buffer;
- var readableBufferLength = readableBuffer.Length;
-
- try
- {
- if (readableBufferLength != 0)
- {
- foreach (var memory in readableBuffer)
- {
- await destination.WriteAsync(memory, cancellationToken);
- }
- }
-
- if (result.IsCompleted)
- {
- return;
- }
- }
- finally
- {
- InnerPipeReader.AdvanceTo(readableBuffer.End);
- }
- }
- }
- }
-}
diff --git a/src/Http/Http/src/StreamPipeReader.cs b/src/Http/Http/src/StreamPipeReader.cs
deleted file mode 100644
index 16e3e0ac32..0000000000
--- a/src/Http/Http/src/StreamPipeReader.cs
+++ /dev/null
@@ -1,393 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System.Buffers;
-using System.Diagnostics;
-using System.Runtime.CompilerServices;
-using System.Runtime.ExceptionServices;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace System.IO.Pipelines
-{
- ///
- /// Implements PipeReader using an underlying stream.
- ///
- public class StreamPipeReader : PipeReader, IDisposable
- {
- private readonly int _bufferSize;
- private readonly int _minimumReadThreshold;
- private readonly MemoryPool _pool;
-
- private CancellationTokenSource _internalTokenSource;
- private bool _isReaderCompleted;
- private bool _isStreamCompleted;
- private ExceptionDispatchInfo _exceptionInfo;
-
- private BufferSegment _readHead;
- private int _readIndex;
-
- private BufferSegment _readTail;
- private long _bufferedBytes;
- private bool _examinedEverything;
- private object _lock = new object();
-
- ///
- /// Creates a new StreamPipeReader.
- ///
- /// The stream to read from.
- public StreamPipeReader(Stream readingStream)
- : this(readingStream, StreamPipeReaderAdapterOptions.DefaultOptions)
- {
- }
-
- ///
- /// Creates a new StreamPipeReader.
- ///
- /// The stream to read from.
- /// The options to use.
- public StreamPipeReader(Stream readingStream, StreamPipeReaderAdapterOptions options)
- {
- InnerStream = readingStream ?? throw new ArgumentNullException(nameof(readingStream));
-
- if (options == null)
- {
- throw new ArgumentNullException(nameof(options));
- }
-
- if (options.MinimumReadThreshold <= 0)
- {
- throw new ArgumentOutOfRangeException(nameof(options.MinimumReadThreshold));
- }
-
- _minimumReadThreshold = Math.Min(options.MinimumReadThreshold, options.MinimumSegmentSize);
- _pool = options.MemoryPool == MemoryPool.Shared ? null : options.MemoryPool;
- _bufferSize = _pool == null ? options.MinimumSegmentSize : Math.Min(options.MinimumSegmentSize, _pool.MaxBufferSize);
- }
-
- ///
- /// Gets the inner stream that is being read from.
- ///
- public Stream InnerStream { get; }
-
- ///
- public override void AdvanceTo(SequencePosition consumed)
- {
- AdvanceTo(consumed, consumed);
- }
-
- private CancellationTokenSource InternalTokenSource
- {
- get
- {
- lock (_lock)
- {
- if (_internalTokenSource == null)
- {
- _internalTokenSource = new CancellationTokenSource();
- }
- return _internalTokenSource;
- }
- }
- set
- {
- _internalTokenSource = value;
- }
- }
-
- ///
- public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
- {
- ThrowIfCompleted();
-
- if (_readHead == null || _readTail == null)
- {
- ThrowHelper.ThrowInvalidOperationException_NoDataRead();
- }
-
- AdvanceTo((BufferSegment)consumed.GetObject(), consumed.GetInteger(), (BufferSegment)examined.GetObject(), examined.GetInteger());
- }
-
- private void AdvanceTo(BufferSegment consumedSegment, int consumedIndex, BufferSegment examinedSegment, int examinedIndex)
- {
- if (consumedSegment == null)
- {
- return;
- }
-
- var returnStart = _readHead;
- var returnEnd = consumedSegment;
-
- var consumedBytes = new ReadOnlySequence(returnStart, _readIndex, consumedSegment, consumedIndex).Length;
-
- _bufferedBytes -= consumedBytes;
-
- Debug.Assert(_bufferedBytes >= 0);
-
- _examinedEverything = false;
-
- if (examinedSegment == _readTail)
- {
- // If we examined everything, we force ReadAsync to actually read from the underlying stream
- // instead of returning a ReadResult from TryRead.
- _examinedEverything = examinedIndex == _readTail.End;
- }
-
- // Three cases here:
- // 1. All data is consumed. If so, we reset _readHead and _readTail to _readTail's original memory owner
- // SetMemory on a IMemoryOwner will reset the internal Memory to be an empty segment
- // 2. A segment is entirely consumed but there is still more data in nextSegments
- // We are allowed to remove an extra segment. by setting returnEnd to be the next block.
- // 3. We are in the middle of a segment.
- // Move _readHead and _readIndex to consumedSegment and index
- if (_bufferedBytes == 0)
- {
- _readTail.SetMemory(_readTail.MemoryOwner);
- _readHead = _readTail;
- returnEnd = _readTail;
- _readIndex = 0;
- }
- else if (consumedIndex == returnEnd.Length)
- {
- var nextBlock = returnEnd.NextSegment;
- _readHead = nextBlock;
- _readIndex = 0;
- returnEnd = nextBlock;
- }
- else
- {
- _readHead = consumedSegment;
- _readIndex = consumedIndex;
- }
-
- // Remove all blocks that are freed (except the last one)
- while (returnStart != returnEnd)
- {
- returnStart.ResetMemory();
- returnStart = returnStart.NextSegment;
- }
- }
-
- ///
- public override void CancelPendingRead()
- {
- InternalTokenSource.Cancel();
- }
-
- ///
- public override void Complete(Exception exception = null)
- {
- if (_isReaderCompleted)
- {
- return;
- }
-
- _isReaderCompleted = true;
- if (exception != null)
- {
- _exceptionInfo = ExceptionDispatchInfo.Capture(exception);
- }
-
- var segment = _readHead;
- while (segment != null)
- {
- segment.ResetMemory();
- segment = segment.NextSegment;
- }
- }
-
- ///
- public override void OnWriterCompleted(Action callback, object state)
- {
- throw new NotSupportedException("OnWriterCompleted is not supported");
- }
-
- ///
- public override async ValueTask ReadAsync(CancellationToken cancellationToken = default)
- {
- // TODO ReadyAsync needs to throw if there are overlapping reads.
- ThrowIfCompleted();
-
- // PERF: store InternalTokenSource locally to avoid querying it twice (which acquires a lock)
- var tokenSource = InternalTokenSource;
- if (TryReadInternal(tokenSource, out var readResult))
- {
- return readResult;
- }
-
- if (_isStreamCompleted)
- {
- return new ReadResult(buffer: default, isCanceled: false, isCompleted: true);
- }
-
- var reg = new CancellationTokenRegistration();
- if (cancellationToken.CanBeCanceled)
- {
- reg = cancellationToken.Register(state => ((StreamPipeReader)state).Cancel(), this);
- }
-
- using (reg)
- {
- var isCanceled = false;
- try
- {
- AllocateReadTail();
-#if NETCOREAPP3_0
- var length = await InnerStream.ReadAsync(_readTail.AvailableMemory.Slice(_readTail.End), tokenSource.Token);
-#elif NETSTANDARD2_0
- if (!MemoryMarshal.TryGetArray(_readTail.AvailableMemory.Slice(_readTail.End), out var arraySegment))
- {
- ThrowHelper.CreateInvalidOperationException_NoArrayFromMemory();
- }
-
- var length = await _readingStream.ReadAsync(arraySegment.Array, arraySegment.Offset, arraySegment.Count, tokenSource.Token);
-#else
-#error Target frameworks need to be updated.
-#endif
- Debug.Assert(length + _readTail.End <= _readTail.AvailableMemory.Length);
-
- _readTail.End += length;
- _bufferedBytes += length;
-
- if (length == 0)
- {
- _isStreamCompleted = true;
- }
- }
- catch (OperationCanceledException)
- {
- ClearCancellationToken();
-
- if (cancellationToken.IsCancellationRequested)
- {
- throw;
- }
-
- isCanceled = true;
- }
-
- return new ReadResult(GetCurrentReadOnlySequence(), isCanceled, IsCompletedOrThrow());
- }
- }
-
- private void ClearCancellationToken()
- {
- lock (_lock)
- {
- _internalTokenSource = null;
- }
- }
-
- private void ThrowIfCompleted()
- {
- if (_isReaderCompleted)
- {
- ThrowHelper.ThrowInvalidOperationException_NoReadingAllowed();
- }
- }
-
- public override bool TryRead(out ReadResult result)
- {
- ThrowIfCompleted();
-
- return TryReadInternal(InternalTokenSource, out result);
- }
-
- private bool TryReadInternal(CancellationTokenSource source, out ReadResult result)
- {
- var isCancellationRequested = source.IsCancellationRequested;
- if (isCancellationRequested || _bufferedBytes > 0 && (!_examinedEverything || _isStreamCompleted))
- {
- // If TryRead/ReadAsync are called and cancellation is requested, we need to make sure memory is allocated for the ReadResult,
- // otherwise if someone calls advance afterward on the ReadResult, it will throw.
- if (isCancellationRequested)
- {
- AllocateReadTail();
-
- ClearCancellationToken();
- }
-
- result = new ReadResult(
- GetCurrentReadOnlySequence(),
- isCanceled: isCancellationRequested,
- IsCompletedOrThrow());
- return true;
- }
-
- result = new ReadResult();
- return false;
- }
-
- private ReadOnlySequence GetCurrentReadOnlySequence()
- {
- return new ReadOnlySequence(_readHead, _readIndex, _readTail, _readTail.End);
- }
-
- private void AllocateReadTail()
- {
- if (_readHead == null)
- {
- Debug.Assert(_readTail == null);
- _readHead = AllocateSegment();
- _readTail = _readHead;
- }
- else if (_readTail.WritableBytes < _minimumReadThreshold)
- {
- CreateNewTailSegment();
- }
- }
-
- private void CreateNewTailSegment()
- {
- BufferSegment nextSegment = AllocateSegment();
- _readTail.SetNext(nextSegment);
- _readTail = nextSegment;
- }
-
- private BufferSegment AllocateSegment()
- {
- var nextSegment = new BufferSegment();
-
- if (_pool is null)
- {
- nextSegment.SetMemory(ArrayPool.Shared.Rent(_bufferSize));
- }
- else
- {
- nextSegment.SetMemory(_pool.Rent(_bufferSize));
- }
-
- return nextSegment;
- }
-
- private void Cancel()
- {
- InternalTokenSource.Cancel();
- }
-
- [MethodImpl(MethodImplOptions.AggressiveInlining)]
- private bool IsCompletedOrThrow()
- {
- if (!_isStreamCompleted)
- {
- return false;
- }
- if (_exceptionInfo != null)
- {
- ThrowLatchedException();
- }
- return true;
- }
-
- [MethodImpl(MethodImplOptions.NoInlining)]
- private void ThrowLatchedException()
- {
- _exceptionInfo.Throw();
- }
-
- public void Dispose()
- {
- Complete();
- }
- }
-}
diff --git a/src/Http/Http/src/StreamPipeReaderOptions.cs b/src/Http/Http/src/StreamPipeReaderOptions.cs
deleted file mode 100644
index 2a932e0c39..0000000000
--- a/src/Http/Http/src/StreamPipeReaderOptions.cs
+++ /dev/null
@@ -1,34 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System;
-using System.Buffers;
-using System.Collections.Generic;
-using System.Text;
-
-namespace System.IO.Pipelines
-{
- public class StreamPipeReaderAdapterOptions
- {
- public static StreamPipeReaderAdapterOptions DefaultOptions = new StreamPipeReaderAdapterOptions();
- public const int DefaultMinimumSegmentSize = 4096;
- public const int DefaultMinimumReadThreshold = 256;
-
- public StreamPipeReaderAdapterOptions()
- {
- }
-
- public StreamPipeReaderAdapterOptions(int minimumSegmentSize, int minimumReadThreshold, MemoryPool memoryPool)
- {
- MinimumSegmentSize = minimumSegmentSize;
- MinimumReadThreshold = minimumReadThreshold;
- MemoryPool = memoryPool;
- }
-
- public int MinimumSegmentSize { get; set; } = DefaultMinimumSegmentSize;
-
- public int MinimumReadThreshold { get; set; } = DefaultMinimumReadThreshold;
-
- public MemoryPool MemoryPool { get; set; } = MemoryPool.Shared;
- }
-}
diff --git a/src/Http/Http/src/StreamPipeWriter.cs b/src/Http/Http/src/StreamPipeWriter.cs
deleted file mode 100644
index 087a2b53b6..0000000000
--- a/src/Http/Http/src/StreamPipeWriter.cs
+++ /dev/null
@@ -1,347 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System.Buffers;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace System.IO.Pipelines
-{
- ///
- /// Implements PipeWriter using a underlying stream.
- ///
- public class StreamPipeWriter : PipeWriter, IDisposable
- {
- private readonly int _minimumSegmentSize;
- private int _bytesWritten;
-
- private List _completedSegments;
- private Memory _currentSegment;
- private object _currentSegmentOwner;
- private MemoryPool _pool;
- private int _position;
-
- private CancellationTokenSource _internalTokenSource;
- private bool _isCompleted;
- private object _lockObject = new object();
-
- private CancellationTokenSource InternalTokenSource
- {
- get
- {
- lock (_lockObject)
- {
- if (_internalTokenSource == null)
- {
- _internalTokenSource = new CancellationTokenSource();
- }
- return _internalTokenSource;
- }
- }
- }
-
- ///
- /// Creates a new StreamPipeWrapper
- ///
- /// The stream to write to
- public StreamPipeWriter(Stream writingStream) : this(writingStream, 4096)
- {
- }
-
- public StreamPipeWriter(Stream writingStream, int minimumSegmentSize, MemoryPool pool = null)
- {
- _minimumSegmentSize = minimumSegmentSize;
- InnerStream = writingStream;
- _pool = pool == MemoryPool.Shared ? null : pool;
- }
-
- ///
- /// Gets the inner stream that is being written to.
- ///
- public Stream InnerStream { get; }
-
- ///
- public override void Advance(int count)
- {
- if (_currentSegment.IsEmpty) // TODO confirm this
- {
- throw new InvalidOperationException("No writing operation. Make sure GetMemory() was called.");
- }
-
- if (count >= 0)
- {
- if (_currentSegment.Length < _position + count)
- {
- throw new InvalidOperationException("Can't advance past buffer size.");
- }
- _bytesWritten += count;
- _position += count;
- }
- }
-
- ///
- public override Memory GetMemory(int sizeHint = 0)
- {
- if (_isCompleted)
- {
- ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
- }
- if (sizeHint < 0)
- {
- ThrowHelper.ThrowArgumentOutOfRangeException(nameof(sizeHint));
- }
-
- EnsureCapacity(sizeHint);
-
- return _currentSegment.Slice(_position);
- }
-
- ///
- public override Span GetSpan(int sizeHint = 0)
- {
- if (_isCompleted)
- {
- ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
- }
- if (sizeHint < 0)
- {
- ThrowHelper.ThrowArgumentOutOfRangeException(nameof(sizeHint));
- }
-
- EnsureCapacity(sizeHint);
-
- return _currentSegment.Span.Slice(_position);
- }
-
- ///
- public override void CancelPendingFlush()
- {
- Cancel();
- }
-
- ///
- public override void Complete(Exception exception = null)
- {
- if (_isCompleted)
- {
- return;
- }
-
- Dispose();
- // We still want to cleanup segments before throwing an exception.
- if (_bytesWritten > 0 && exception == null)
- {
- ThrowHelper.ThrowInvalidOperationException_DataNotAllFlushed();
- }
- }
-
- ///
- public override void OnReaderCompleted(Action callback, object state)
- {
- throw new NotSupportedException("OnReaderCompleted isn't supported in StreamPipeWrapper.");
- }
-
- ///
- public override ValueTask FlushAsync(CancellationToken cancellationToken = default)
- {
- if (_bytesWritten == 0)
- {
- return new ValueTask(new FlushResult(isCanceled: false, _isCompleted));
- }
-
- return FlushAsyncInternal(cancellationToken);
- }
-
- private void Cancel()
- {
- InternalTokenSource.Cancel();
- }
-
- private async ValueTask FlushAsyncInternal(CancellationToken cancellationToken = default)
- {
- // Write all completed segments and whatever remains in the current segment
- // and flush the result.
- var reg = new CancellationTokenRegistration();
- if (cancellationToken.CanBeCanceled)
- {
- reg = cancellationToken.Register(state => ((StreamPipeWriter)state).Cancel(), this);
- }
- using (reg)
- {
- var localToken = InternalTokenSource.Token;
- try
- {
- if (_completedSegments != null && _completedSegments.Count > 0)
- {
- var count = _completedSegments.Count;
- for (var i = 0; i < count; i++)
- {
- var segment = _completedSegments[0];
-#if NETCOREAPP3_0
- await InnerStream.WriteAsync(segment.Buffer.Slice(0, segment.Length), localToken);
-#elif NETSTANDARD2_0
- MemoryMarshal.TryGetArray(segment.Buffer, out var arraySegment);
- await InnerStream.WriteAsync(arraySegment.Array, 0, segment.Length, localToken);
-#else
-#error Target frameworks need to be updated.
-#endif
- _bytesWritten -= segment.Length;
- segment.Return();
- _completedSegments.RemoveAt(0);
- }
- }
-
- if (!_currentSegment.IsEmpty)
- {
-#if NETCOREAPP3_0
- await InnerStream.WriteAsync(_currentSegment.Slice(0, _position), localToken);
-#elif NETSTANDARD2_0
- MemoryMarshal.TryGetArray(_currentSegment, out var arraySegment);
- await InnerStream.WriteAsync(arraySegment.Array, 0, _position, localToken);
-#else
-#error Target frameworks need to be updated.
-#endif
- _bytesWritten -= _position;
- _position = 0;
- }
-
- await InnerStream.FlushAsync(localToken);
-
- return new FlushResult(isCanceled: false, _isCompleted);
- }
- catch (OperationCanceledException)
- {
- // Remove the cancellation token such that the next time Flush is called
- // A new CTS is created.
- lock (_lockObject)
- {
- _internalTokenSource = null;
- }
-
- if (cancellationToken.IsCancellationRequested)
- {
- throw;
- }
-
- // Catch any cancellation and translate it into setting isCanceled = true
- return new FlushResult(isCanceled: true, _isCompleted);
- }
- }
- }
-
- private void EnsureCapacity(int sizeHint)
- {
- // This does the Right Thing. It only subtracts _position from the current segment length if it's non-null.
- // If _currentSegment is null, it returns 0.
- var remainingSize = _currentSegment.Length - _position;
-
- // If the sizeHint is 0, any capacity will do
- // Otherwise, the buffer must have enough space for the entire size hint, or we need to add a segment.
- if ((sizeHint == 0 && remainingSize > 0) || (sizeHint > 0 && remainingSize >= sizeHint))
- {
- // We have capacity in the current segment
- return;
- }
-
- AddSegment(sizeHint);
- }
-
- private void AddSegment(int sizeHint = 0)
- {
- if (_currentSegment.Length != 0)
- {
- // We're adding a segment to the list
- if (_completedSegments == null)
- {
- _completedSegments = new List();
- }
-
- // Position might be less than the segment length if there wasn't enough space to satisfy the sizeHint when
- // GetMemory was called. In that case we'll take the current segment and call it "completed", but need to
- // ignore any empty space in it.
- _completedSegments.Add(new CompletedBuffer(_currentSegmentOwner, _currentSegment, _position));
- }
-
- if (_pool is null)
- {
- _currentSegment = ArrayPool.Shared.Rent(Math.Max(sizeHint, _minimumSegmentSize));
- _currentSegmentOwner = _currentSegment;
- }
- else if (sizeHint <= _pool.MaxBufferSize)
- {
- // Get a new buffer using the minimum segment size, unless the size hint is larger than a single segment.
- // Also, the size cannot be larger than the MaxBufferSize of the MemoryPool
- var owner = _pool.Rent(Math.Clamp(sizeHint, _minimumSegmentSize, _pool.MaxBufferSize));
- _currentSegment = owner.Memory;
- _currentSegmentOwner = owner;
- }
- else
- {
- _currentSegment = new byte[sizeHint];
- }
-
- _position = 0;
- }
-
- public void Dispose()
- {
- if (_isCompleted)
- {
- return;
- }
-
- _isCompleted = true;
-
- _internalTokenSource?.Dispose();
-
- if (_completedSegments != null)
- {
- foreach (var segment in _completedSegments)
- {
- segment.Return();
- }
- }
-
- DisposeOwner(_currentSegmentOwner);
- }
-
- private static void DisposeOwner(object owner)
- {
- if (owner is IMemoryOwner memoryOwner)
- {
- memoryOwner.Dispose();
- }
- else if (owner is byte[] array)
- {
- ArrayPool.Shared.Return(array);
- }
- }
-
- ///
- /// Holds a byte[] from the pool and a size value. Basically a Memory but guaranteed to be backed by an ArrayPool byte[], so that we know we can return it.
- ///
- private readonly struct CompletedBuffer
- {
- private readonly object _memoryOwner;
-
- public Memory Buffer { get; }
- public int Length { get; }
-
- public ReadOnlySpan Span => Buffer.Span;
-
- public CompletedBuffer(object owner, Memory buffer, int length)
- {
- _memoryOwner = owner;
-
- Buffer = buffer;
- Length = length;
- }
-
- public void Return()
- {
- DisposeOwner(_memoryOwner);
- }
- }
- }
-}
diff --git a/src/Http/Http/src/ThrowHelper.cs b/src/Http/Http/src/ThrowHelper.cs
deleted file mode 100644
index 0745b9bf65..0000000000
--- a/src/Http/Http/src/ThrowHelper.cs
+++ /dev/null
@@ -1,50 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System.Runtime.CompilerServices;
-
-namespace System.IO.Pipelines
-{
- internal static class ThrowHelper
- {
- public static void ThrowInvalidOperationException_NoReadingAllowed() => throw CreateInvalidOperationException_NoReadingAllowed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_NoReadingAllowed() => new InvalidOperationException("Reading is not allowed after reader was completed.");
-
- public static void ThrowInvalidOperationException_NoArrayFromMemory() => throw CreateInvalidOperationException_NoArrayFromMemory();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_NoArrayFromMemory() => new InvalidOperationException("Could not get byte[] from Memory.");
-
- public static void ThrowInvalidOperationException_NoDataRead() => throw CreateInvalidOperationException_NoDataRead();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_NoDataRead() => new InvalidOperationException("No data has been read into the StreamPipeReader.");
-
- public static void ThrowInvalidOperationException_SynchronousReadsDisallowed() => throw CreateInvalidOperationException_SynchronousReadsDisallowed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_SynchronousReadsDisallowed() => new InvalidOperationException("Synchronous operations are disallowed. Call ReadAsync or set allowSynchronousIO to true instead.");
-
- public static void ThrowInvalidOperationException_SynchronousWritesDisallowed() => throw CreateInvalidOperationException_SynchronousWritesDisallowed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_SynchronousWritesDisallowed() => new InvalidOperationException("Synchronous operations are disallowed. Call WriteAsync or set allowSynchronousIO to true instead.");
-
- public static void ThrowInvalidOperationException_SynchronousFlushesDisallowed() => throw CreateInvalidOperationException_SynchronousFlushesDisallowed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_SynchronousFlushesDisallowed() => new InvalidOperationException("Synchronous operations are disallowed. Call FlushAsync or set allowSynchronousIO to true instead.");
-
- public static void ThrowInvalidOperationException_DataNotAllFlushed() => throw CreateInvalidOperationException_DataNotAllFlushed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_DataNotAllFlushed() => new InvalidOperationException("Complete called without flushing the StreamPipeWriter. Call FlushAsync() before calling Complete().");
-
- public static void ThrowInvalidOperationException_NoWritingAllowed() => throw CreateInvalidOperationException_NoWritingAllowed();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateInvalidOperationException_NoWritingAllowed() => new InvalidOperationException("Writing is not allowed after writer was completed.");
-
- public static void ThrowArgumentOutOfRangeException(string argument) => throw CreateArgumentOutOfRangeException(argument);
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateArgumentOutOfRangeException(string argument) => new ArgumentOutOfRangeException(argument);
-
- public static void ThrowOperationCanceledException_ReadCanceled() => throw CreateOperationCanceledException_ReadCanceled();
- [MethodImpl(MethodImplOptions.NoInlining)]
- public static Exception CreateOperationCanceledException_ReadCanceled() => new OperationCanceledException("Read was canceled on underlying PipeReader.");
- }
-}
diff --git a/src/Http/Http/src/WriteOnlyPipeStream.cs b/src/Http/Http/src/WriteOnlyPipeStream.cs
deleted file mode 100644
index 1d8d23328b..0000000000
--- a/src/Http/Http/src/WriteOnlyPipeStream.cs
+++ /dev/null
@@ -1,169 +0,0 @@
-// Copyright (c) .NET Foundation. All rights reserved.
-// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
-
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore.Internal;
-
-namespace System.IO.Pipelines
-{
- ///
- /// Represents a WriteOnlyStream backed by a PipeWriter
- ///
- public class WriteOnlyPipeStream : Stream
- {
- private bool _allowSynchronousIO = true;
-
- ///
- /// Creates a new WriteOnlyStream
- ///
- /// The PipeWriter to write to.
- public WriteOnlyPipeStream(PipeWriter pipeWriter) :
- this(pipeWriter, allowSynchronousIO: true)
- {
- }
-
- ///
- /// Creates a new WriteOnlyStream
- ///
- /// The PipeWriter to write to.
- /// Whether synchronous IO is allowed.
- public WriteOnlyPipeStream(PipeWriter pipeWriter, bool allowSynchronousIO)
- {
- InnerPipeWriter = pipeWriter;
- _allowSynchronousIO = allowSynchronousIO;
- }
-
- ///
- public override bool CanSeek => false;
-
- ///
- public override bool CanRead => false;
-
- ///
- public override bool CanWrite => true;
-
- ///
- public override long Length => throw new NotSupportedException();
-
- ///
- public override long Position
- {
- get => throw new NotSupportedException();
- set => throw new NotSupportedException();
- }
-
- ///
- public override int ReadTimeout
- {
- get => throw new NotSupportedException();
- set => throw new NotSupportedException();
- }
-
- public PipeWriter InnerPipeWriter { get; }
-
- ///
- public override int Read(byte[] buffer, int offset, int count)
- => throw new NotSupportedException();
-
- ///
- public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- => throw new NotSupportedException();
-
- ///
- public override void Flush()
- {
- if (!_allowSynchronousIO)
- {
- ThrowHelper.ThrowInvalidOperationException_SynchronousFlushesDisallowed();
- }
-
- FlushAsync(default).GetAwaiter().GetResult();
- }
-
- ///
- public override Task FlushAsync(CancellationToken cancellationToken)
- {
- return InnerPipeWriter.FlushAsync(cancellationToken).GetAsTask();
- }
-
- ///
- public override long Seek(long offset, SeekOrigin origin)
- {
- throw new NotSupportedException();
- }
-
- ///
- public override void SetLength(long value)
- {
- throw new NotSupportedException();
- }
-
- ///
- public override void Write(byte[] buffer, int offset, int count)
- {
- if (!_allowSynchronousIO)
- {
- ThrowHelper.ThrowInvalidOperationException_SynchronousWritesDisallowed();
- }
- WriteAsync(buffer, offset, count, default).GetAwaiter().GetResult();
- }
-
- ///
- public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
- {
- var task = WriteAsync(buffer, offset, count, default, state);
- if (callback != null)
- {
- task.ContinueWith(t => callback.Invoke(t));
- }
- return task;
- }
-
- ///
- public override void EndWrite(IAsyncResult asyncResult)
- {
- ((Task