Pipelines in corefx (#1501)

build/dependencies.props

@@ -22,16 +22,16 @@
     <MicrosoftAspNetCoreHttpPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreHttpPackageVersion>
     <MicrosoftAspNetCoreIdentityEntityFrameworkCorePackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreIdentityEntityFrameworkCorePackageVersion>
     <MicrosoftAspNetCoreMvcPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreMvcPackageVersion>
-    <MicrosoftAspNetCoreProtocolsAbstractionsPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreProtocolsAbstractionsPackageVersion>
+    <MicrosoftAspNetCoreProtocolsAbstractionsPackageVersion>2.1.0-preview2-t000</MicrosoftAspNetCoreProtocolsAbstractionsPackageVersion>
     <MicrosoftAspNetCoreRoutingPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreRoutingPackageVersion>
     <MicrosoftAspNetCoreServerIISIntegrationPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreServerIISIntegrationPackageVersion>
     <MicrosoftAspNetCoreServerIntegrationTestingPackageVersion>0.5.0-preview2-30187</MicrosoftAspNetCoreServerIntegrationTestingPackageVersion>
-    <MicrosoftAspNetCoreServerKestrelPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreServerKestrelPackageVersion>
+    <MicrosoftAspNetCoreServerKestrelPackageVersion>2.1.0-preview2-t000</MicrosoftAspNetCoreServerKestrelPackageVersion>
     <MicrosoftAspNetCoreStaticFilesPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreStaticFilesPackageVersion>
     <MicrosoftAspNetCoreTestHostPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreTestHostPackageVersion>
     <MicrosoftAspNetCoreTestingPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreTestingPackageVersion>
     <MicrosoftAspNetCoreWebSocketsPackageVersion>2.1.0-preview2-30187</MicrosoftAspNetCoreWebSocketsPackageVersion>
-    <MicrosoftCSharpPackageVersion>4.5.0-preview2-26130-01</MicrosoftCSharpPackageVersion>
+    <MicrosoftCSharpPackageVersion>4.5.0-preview2-26224-02</MicrosoftCSharpPackageVersion>
     <MicrosoftEntityFrameworkCoreDesignPackageVersion>2.1.0-preview2-30187</MicrosoftEntityFrameworkCoreDesignPackageVersion>
     <MicrosoftEntityFrameworkCoreSqlServerPackageVersion>2.1.0-preview2-30187</MicrosoftEntityFrameworkCoreSqlServerPackageVersion>
     <MicrosoftEntityFrameworkCoreToolsPackageVersion>2.1.0-preview2-30187</MicrosoftEntityFrameworkCoreToolsPackageVersion>
@@ -53,22 +53,21 @@
     <MicrosoftExtensionsSecurityHelperSourcesPackageVersion>2.1.0-preview2-30187</MicrosoftExtensionsSecurityHelperSourcesPackageVersion>
     <MicrosoftExtensionsValueStopwatchSourcesPackageVersion>2.1.0-preview2-30187</MicrosoftExtensionsValueStopwatchSourcesPackageVersion>
     <MicrosoftNETCoreApp20PackageVersion>2.0.0</MicrosoftNETCoreApp20PackageVersion>
-    <MicrosoftNETCoreApp21PackageVersion>2.1.0-preview2-26130-04</MicrosoftNETCoreApp21PackageVersion>
+    <MicrosoftNETCoreApp21PackageVersion>2.1.0-preview2-26225-03</MicrosoftNETCoreApp21PackageVersion>
     <MicrosoftNETTestSdkPackageVersion>15.6.0</MicrosoftNETTestSdkPackageVersion>
     <MoqPackageVersion>4.7.49</MoqPackageVersion>
     <MsgPackCliPackageVersion>1.0.0-rc</MsgPackCliPackageVersion>
     <NewtonsoftJsonPackageVersion>10.0.1</NewtonsoftJsonPackageVersion>
     <StackExchangeRedisStrongNamePackageVersion>1.2.4</StackExchangeRedisStrongNamePackageVersion>
-    <SystemBuffersPackageVersion>4.5.0-preview2-26130-01</SystemBuffersPackageVersion>
+    <SystemBuffersPackageVersion>4.5.0-preview2-26224-02</SystemBuffersPackageVersion>
-    <SystemBuffersPrimitivesPackageVersion>0.1.0-preview2-180130-1</SystemBuffersPrimitivesPackageVersion>
+    <SystemIOPipelinesPackageVersion>4.5.0-preview2-26224-02</SystemIOPipelinesPackageVersion>
-    <SystemIOPipelinesPackageVersion>0.1.0-preview2-180130-1</SystemIOPipelinesPackageVersion>
+    <SystemMemoryPackageVersion>4.5.0-preview2-26224-02</SystemMemoryPackageVersion>
-    <SystemMemoryPackageVersion>4.5.0-preview2-26130-01</SystemMemoryPackageVersion>
+    <SystemNumericsVectorsPackageVersion>4.5.0-preview2-26224-02</SystemNumericsVectorsPackageVersion>
-    <SystemNumericsVectorsPackageVersion>4.5.0-preview2-26130-01</SystemNumericsVectorsPackageVersion>
     <SystemReactiveLinqPackageVersion>3.1.1</SystemReactiveLinqPackageVersion>
     <SystemReflectionEmitPackageVersion>4.3.0</SystemReflectionEmitPackageVersion>
-    <SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview2-26130-01</SystemRuntimeCompilerServicesUnsafePackageVersion>
+    <SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview2-26224-02</SystemRuntimeCompilerServicesUnsafePackageVersion>
-    <SystemThreadingChannelsPackageVersion>4.5.0-preview2-26130-01</SystemThreadingChannelsPackageVersion>
+    <SystemThreadingChannelsPackageVersion>4.5.0-preview2-26224-02</SystemThreadingChannelsPackageVersion>
-    <SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview2-26130-01</SystemThreadingTasksExtensionsPackageVersion>
+    <SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview2-26224-02</SystemThreadingTasksExtensionsPackageVersion>
     <XunitPackageVersion>2.3.1</XunitPackageVersion>
     <XunitRunnerVisualStudioPackageVersion>2.4.0-beta.1.build3945</XunitRunnerVisualStudioPackageVersion>
   </PropertyGroup>

client-ts/FunctionalTests/EchoEndPoint.cs

@@ -1,7 +1,7 @@
 // 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.Pipelines;
+using System.Buffers;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Protocols;
 using Microsoft.AspNetCore.Sockets;

samples/SocialWeather/SocialWeatherEndPoint.cs

@@ -1,6 +1,7 @@
 // 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.IO;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Protocols;

samples/SocketsSample/EndPoints/MessagesEndPoint.cs

@@ -1,6 +1,7 @@
 // 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.IO.Pipelines;
 using System.Text;
@@ -64,10 +65,11 @@ namespace SocketsSample.EndPoints
         private Task Broadcast(byte[] payload)
         {
             var tasks = new List<Task>(Connections.Count);
+            async Task<FlushResult> ToTask(PipeAwaiter<FlushResult> awaiter) => await awaiter;
 
             foreach (var c in Connections)
             {
-                tasks.Add(c.Transport.Output.WriteAsync(payload));
+                tasks.Add(ToTask(c.Transport.Output.WriteAsync(payload)));
             }
 
             return Task.WhenAll(tasks);

src/Common/StreamExtensions.cs

@@ -11,7 +11,7 @@ namespace System.IO
 {
     internal static class StreamExtensions
     {
-        public static async Task WriteAsync(this Stream stream, ReadOnlyBuffer<byte> buffer, CancellationToken cancellationToken = default)
+        public static async Task WriteAsync(this Stream stream, ReadOnlySequence<byte> buffer, CancellationToken cancellationToken = default)
         {
             // REVIEW: Should we special case IsSingleSegment here?
             foreach (var segment in buffer)

src/Common/WebSocketExtensions.cs

@@ -14,7 +14,7 @@ namespace System.Net.WebSockets
 {
     internal static class WebSocketExtensions
     {
-        public static Task SendAsync(this WebSocket webSocket, ReadOnlyBuffer<byte> buffer, WebSocketMessageType webSocketMessageType, CancellationToken cancellationToken = default)
+        public static Task SendAsync(this WebSocket webSocket, ReadOnlySequence<byte> buffer, WebSocketMessageType webSocketMessageType, CancellationToken cancellationToken = default)
         {
             // TODO: Consider chunking writes here if we get a multi segment buffer
 #if NETCOREAPP2_1

src/Microsoft.AspNetCore.SignalR.Common/Internal/HubProtocolReaderWriter.cs

@@ -22,7 +22,7 @@ namespace Microsoft.AspNetCore.SignalR.Internal
             _dataEncoder = dataEncoder;
         }
 
-        public bool ReadMessages(ReadOnlyBuffer<byte> buffer, IInvocationBinder binder, out IList<HubMessage> messages, out SequencePosition consumed, out SequencePosition examined)
+        public bool ReadMessages(ReadOnlySequence<byte> buffer, IInvocationBinder binder, out IList<HubMessage> messages, out SequencePosition consumed, out SequencePosition examined)
         {
             // TODO: Fix this implementation to be incremental
             consumed = buffer.End;

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/NegotiationProtocol.cs

@@ -56,7 +56,7 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
             return true;
         }
 
-        public static bool TryParseMessage(ReadOnlyBuffer<byte> buffer, out NegotiationMessage negotiationMessage, out SequencePosition consumed, out SequencePosition examined)
+        public static bool TryParseMessage(ReadOnlySequence<byte> buffer, out NegotiationMessage negotiationMessage, out SequencePosition consumed, out SequencePosition examined)
         {
             var separator = buffer.PositionOf(TextMessageFormatter.RecordSeparator);
             if (separator == null)

src/Microsoft.AspNetCore.SignalR.Common/Microsoft.AspNetCore.SignalR.Common.csproj

@@ -11,7 +11,6 @@
     <PackageReference Include="Newtonsoft.Json" Version="$(NewtonsoftJsonPackageVersion)" />
     <PackageReference Include="System.Memory" Version="$(SystemMemoryPackageVersion)" />
     <PackageReference Include="System.Buffers" Version="$(SystemBuffersPackageVersion)" />
-    <PackageReference Include="System.Buffers.Primitives" Version="$(SystemBuffersPrimitivesPackageVersion)" />
     <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="$(SystemRuntimeCompilerServicesUnsafePackageVersion)" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="$(MicrosoftExtensionsOptionsPackageVersion)" />
   </ItemGroup>

src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs

@@ -2,6 +2,7 @@
 // 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.IO;
 using System.IO.Pipelines;

src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/StreamExtensions.cs

@@ -38,12 +38,12 @@ namespace System.IO.Pipelines
         }
 
         /// <summary>
-        /// Copies a <see cref="ReadOnlyBuffer{Byte}"/> to a <see cref="Stream"/> asynchronously
+        /// Copies a <see cref="ReadOnlySequence{Byte}"/> to a <see cref="Stream"/> asynchronously
         /// </summary>
-        /// <param name="buffer">The <see cref="ReadOnlyBuffer{Byte}"/> to copy</param>
+        /// <param name="buffer">The <see cref="ReadOnlySequence{Byte}"/> to copy</param>
         /// <param name="stream">The target <see cref="Stream"/></param>
         /// <returns></returns>
-        public static Task CopyToAsync(this ReadOnlyBuffer<byte> buffer, Stream stream)
+        public static Task CopyToAsync(this ReadOnlySequence<byte> buffer, Stream stream)
         {
             if (buffer.IsSingleSegment)
             {
@@ -53,7 +53,7 @@ namespace System.IO.Pipelines
             return CopyMultipleToStreamAsync(buffer, stream);
         }
 
-        private static async Task CopyMultipleToStreamAsync(this ReadOnlyBuffer<byte> buffer, Stream stream)
+        private static async Task CopyMultipleToStreamAsync(this ReadOnlySequence<byte> buffer, Stream stream)
         {
             foreach (var memory in buffer)
             {

src/Microsoft.AspNetCore.Sockets.Client.Http/SendUtils.cs

@@ -42,7 +42,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
                             var request = new HttpRequestMessage(HttpMethod.Post, sendUrl);
                             PrepareHttpRequest(request, httpOptions);
 
-                            request.Content = new ReadOnlyBufferContent(buffer);
+                            request.Content = new ReadOnlySequenceContent(buffer);
 
                             var response = await httpClient.SendAsync(request, transportCts.Token);
                             response.EnsureSuccessStatusCode();
@@ -99,11 +99,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
             }
         }
 
-        private class ReadOnlyBufferContent : HttpContent
+        private class ReadOnlySequenceContent : HttpContent
         {
-            private readonly ReadOnlyBuffer<byte> _buffer;
+            private readonly ReadOnlySequence<byte> _buffer;
 
-            public ReadOnlyBufferContent(ReadOnlyBuffer<byte> buffer)
+            public ReadOnlySequenceContent(ReadOnlySequence<byte> buffer)
             {
                 _buffer = buffer;
             }

src/Microsoft.AspNetCore.Sockets.Client.Http/ServerSentEventsMessageParser.cs

@@ -3,10 +3,7 @@
 
 using System;
 using System.Buffers;
-using System.Collections;
 using System.Collections.Generic;
-using System.Collections.Sequences;
-using System.IO.Pipelines;
 using System.Runtime.CompilerServices;
 using System.Text;
 
@@ -25,7 +22,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
         private InternalParseState _internalParserState = InternalParseState.ReadMessagePayload;
         private List<byte[]> _data = new List<byte[]>();
 
-        public ParseResult ParseMessage(ReadOnlyBuffer<byte> buffer, out SequencePosition consumed, out SequencePosition examined, out byte[] message)
+        public ParseResult ParseMessage(ReadOnlySequence<byte> buffer, out SequencePosition consumed, out SequencePosition examined, out byte[] message)
         {
             consumed = buffer.Start;
             examined = buffer.End;
@@ -149,7 +146,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        private ReadOnlySpan<byte> ConvertBufferToSpan(ReadOnlyBuffer<byte> buffer)
+        private ReadOnlySpan<byte> ConvertBufferToSpan(ReadOnlySequence<byte> buffer)
         {
             if (buffer.IsSingleSegment)
             {

src/Microsoft.AspNetCore.Sockets.Client.Http/ServerSentEventsTransport.cs

@@ -93,7 +93,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
                     {
                         var result = await pipelineReader.ReadAsync();
                         var input = result.Buffer;
-                        if (result.IsCancelled || (input.IsEmpty && result.IsCompleted))
+                        if (result.IsCanceled || (input.IsEmpty && result.IsCompleted))
                         {
                             _logger.EventStreamEnded();
                             break;

src/Microsoft.AspNetCore.Sockets.Client.Http/WebSocketsTransport.cs

@@ -177,7 +177,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
 
                         // We canceled in the middle of applying back pressure
                         // or if the consumer is done
-                        if (flushResult.IsCancelled || flushResult.IsCompleted)
+                        if (flushResult.IsCanceled || flushResult.IsCompleted)
                         {
                             break;
                         }
@@ -228,7 +228,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
 
                     try
                     {
-                        if (result.IsCancelled)
+                        if (result.IsCanceled)
                         {
                             break;
                         }

src/Microsoft.AspNetCore.Sockets.Http/Internal/Transports/ServerSentEventsMessageFormatter.cs

@@ -17,7 +17,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
 
         private const byte LineFeed = (byte)'\n';
 
-        public static async Task WriteMessageAsync(ReadOnlyBuffer<byte> payload, Stream output)
+        public static async Task WriteMessageAsync(ReadOnlySequence<byte> payload, Stream output)
         {
             var ms = new MemoryStream();
 

src/Microsoft.AspNetCore.Sockets.Http/Internal/Transports/WebSocketsTransport.cs

@@ -164,7 +164,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Transports
 
                         // We canceled in the middle of applying back pressure
                         // or if the consumer is done
-                        if (flushResult.IsCancelled || flushResult.IsCompleted)
+                        if (flushResult.IsCanceled || flushResult.IsCompleted)
                         {
                             break;
                         }
@@ -208,7 +208,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Transports
 
                     try
                     {
-                        if (result.IsCancelled)
+                        if (result.IsCanceled)
                         {
                             break;
                         }

test/Microsoft.AspNetCore.SignalR.Client.Tests/ServerSentEventsParserTests.cs

@@ -3,9 +3,7 @@
 
 using System;
 using System.Buffers;
-using System.Collections;
 using System.Collections.Generic;
-using System.Collections.Sequences;
 using System.IO.Pipelines;
 using System.Text;
 using System.Threading.Tasks;
@@ -33,7 +31,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         public void ParseSSEMessageSuccessCases(string encodedMessage, string expectedMessage)
         {
             var buffer = Encoding.UTF8.GetBytes(encodedMessage);
-            var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
+            var readableBuffer = new ReadOnlySequence<byte>(buffer);
             var parser = new ServerSentEventsMessageParser();
 
             var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);
@@ -59,7 +57,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         public void ParseSSEMessageFailureCases(string encodedMessage, string expectedExceptionMessage)
         {
             var buffer = Encoding.UTF8.GetBytes(encodedMessage);
-            var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
+            var readableBuffer = new ReadOnlySequence<byte>(buffer);
             var parser = new ServerSentEventsMessageParser();
 
             var ex = Assert.Throws<FormatException>(() => { parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); });
@@ -87,7 +85,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         public void ParseSSEMessageIncompleteParseResult(string encodedMessage)
         {
             var buffer = Encoding.UTF8.GetBytes(encodedMessage);
-            var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
+            var readableBuffer = new ReadOnlySequence<byte>(buffer);
             var parser = new ServerSentEventsMessageParser();
 
             var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);
@@ -110,9 +108,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         public async Task ParseMessageAcrossMultipleReadsSuccess(string[] messageParts, string expectedMessage)
         {
             var parser = new ServerSentEventsMessageParser();
-            using (var pool = new MemoryPool())
+            var pipe = new Pipe();
-            {
-                var pipe = new Pipe(new PipeOptions(pool));
 
                 byte[] message = null;
                 SequencePosition consumed = default, examined = default;
@@ -140,7 +136,6 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
                 var resultMessage = Encoding.UTF8.GetString(message);
                 Assert.Equal(expectedMessage, resultMessage);
         }
-        }
 
         [Theory]
         [InlineData("data: T", "\n", "Unexpected '\n' in message. A '\n' character can only be used as part of the newline sequence '\r\n'")]
@@ -155,9 +150,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         [InlineData("data: B\r\ndata: SGVs", "bG8sIFdvcmxk\r\n\n\n", "There was an error in the frame format")]
         public async Task ParseMessageAcrossMultipleReadsFailure(string encodedMessagePart1, string encodedMessagePart2, string expectedMessage)
         {
-            using (var pool = new MemoryPool())
+            var pipe = new Pipe();
-            {
-                var pipe = new Pipe(new PipeOptions(pool));
 
                 // Read the first part of the message
                 await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart1));
@@ -177,15 +170,12 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
                 var ex = Assert.Throws<FormatException>(() => parser.ParseMessage(result.Buffer, out consumed, out examined, out buffer));
                 Assert.Equal(expectedMessage, ex.Message);
         }
-        }
 
         [Theory]
         [InlineData("data: foo\r\n\r\n", "data: bar\r\n\r\n")]
         public async Task ParseMultipleMessagesText(string message1, string message2)
         {
-            using (var pool = new MemoryPool())
+            var pipe = new Pipe();
-            {
-                var pipe = new Pipe(new PipeOptions(pool));
 
                 // Read the first part of the message
                 await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(message1 + message2));
@@ -208,7 +198,6 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
                 Assert.Equal("bar", Encoding.UTF8.GetString(message));
                 pipe.Reader.AdvanceTo(consumed, examined);
         }
-        }
 
         public static IEnumerable<object[]> MultilineMessages
         {
@@ -224,7 +213,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         public void ParseMessagesWithMultipleDataLines(string encodedMessage, string expectedMessage)
         {
             var buffer = Encoding.UTF8.GetBytes(encodedMessage);
-            var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
+            var readableBuffer = new ReadOnlySequence<byte>(buffer);
             var parser = new ServerSentEventsMessageParser();
 
             var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);

test/Microsoft.AspNetCore.SignalR.Tests.Utils/TaskExtensions.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System.Diagnostics;
+using System.IO.Pipelines;
 using System.Runtime.CompilerServices;
 
 namespace System.Threading.Tasks
@@ -92,5 +93,10 @@ namespace System.Threading.Tasks
                 return "Operation timed out";
             }
         }
+
+        public static async Task<T> AsTask<T>(this PipeAwaiter<T> awaiter)
+        {
+            return await awaiter;
+        }
     }
 }

test/Microsoft.AspNetCore.SignalR.Tests/EchoEndPoint.cs

@@ -1,6 +1,7 @@
 // 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.IO.Pipelines;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Protocols;

test/Microsoft.AspNetCore.SignalR.Tests/WebSocketsTransportTests.cs

@@ -2,6 +2,7 @@
 // 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.IO.Pipelines;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Sockets;

test/Microsoft.AspNetCore.Sockets.Tests/HttpConnectionDispatcherTests.cs

@@ -2,6 +2,7 @@
 // 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.IO;
 using System.IO.Pipelines;
@@ -86,7 +87,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                 // Reading here puts us below the threshold
                 await connection.Transport.Input.ConsumeAsync(5);
 
-                await writeTask.OrTimeout();
+                await writeTask.AsTask().OrTimeout();
             }
         }
 
@@ -987,7 +988,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                 context.User = new ClaimsPrincipal(new ClaimsIdentity(new[] { new Claim(ClaimTypes.NameIdentifier, "name") }));
 
                 var endPointTask = dispatcher.ExecuteAsync(context, options, app);
-                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).OrTimeout();
+                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).AsTask().OrTimeout();
 
                 await endPointTask.OrTimeout();
 
@@ -1068,7 +1069,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             }));
 
                 var endPointTask = dispatcher.ExecuteAsync(context, options, app);
-                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).OrTimeout();
+                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).AsTask().OrTimeout();
 
                 await endPointTask.OrTimeout();
 
@@ -1125,7 +1126,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                 context.User = new ClaimsPrincipal(new ClaimsIdentity(new[] { new Claim(ClaimTypes.NameIdentifier, "name") }));
 
                 var endPointTask = dispatcher.ExecuteAsync(context, options, app);
-                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).OrTimeout();
+                await connection.Transport.Output.WriteAsync(Encoding.UTF8.GetBytes("Hello, World")).AsTask().OrTimeout();
 
                 await endPointTask.OrTimeout();
 

test/Microsoft.AspNetCore.Sockets.Tests/WebSocketsTests.cs

@@ -2,6 +2,7 @@
 // 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.IO.Pipelines;
 using System.Net.WebSockets;
 using System.Text;