tidy up status codes and transport conflicts (#306)

SignalR.sln

@@ -47,7 +47,7 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WebSocketSample", "samples\
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AspNetCore.Sockets.Client", "src\Microsoft.AspNetCore.Sockets.Client\Microsoft.AspNetCore.Sockets.Client.csproj", "{623FD372-36DE-41A9-A564-F6040D570DBD}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AspNetCore.Client.Tests", "test\Microsoft.AspNetCore.Sockets.Client.Tests\Microsoft.AspNetCore.Client.Tests.csproj", "{B19C15A5-F5EA-4CA7-936B-1166ABEE35C4}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AspNetCore.SignalR.Client.Tests", "test\Microsoft.AspNetCore.SignalR.Client.Tests\Microsoft.AspNetCore.SignalR.Client.Tests.csproj", "{B19C15A5-F5EA-4CA7-936B-1166ABEE35C4}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AspNetCore.SignalR.Common", "src\Microsoft.AspNetCore.SignalR.Common\Microsoft.AspNetCore.SignalR.Common.csproj", "{E37324FF-6BAF-4243-BA80-7C024CF5F29D}"
 EndProject

src/Microsoft.AspNetCore.Sockets/Transports/LongPollingTransport.cs

@@ -49,6 +49,8 @@ namespace Microsoft.AspNetCore.Sockets.Transports
 
                 output.Append(MessageFormatter.GetFormatIndicator(messageFormat));
 
+                // We're intentionally not checking cancellation here because we need to drain messages we've got so far,
+                // but it's too late to emit the 204 required by being cancelled.
                 while (_application.TryRead(out var message))
                 {
                     _logger.LogDebug("Writing {0} byte message to response", message.Payload.Length);

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

@@ -10,6 +10,7 @@ using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Http.Internal;
 using Microsoft.AspNetCore.SignalR.Tests.Common;
 using Microsoft.AspNetCore.Sockets.Internal;
+using Microsoft.AspNetCore.WebSockets.Internal;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Primitives;
@@ -177,16 +178,18 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             Assert.False(exists);
         }
 
-        [Fact]
+        [Theory]
-        public async Task RequestToActiveConnectionId409ForStreamingTransports()
+        [InlineData("/ws", true)]
+        [InlineData("/sse", false)]
+        public async Task RequestToActiveConnectionId409ForStreamingTransports(string path, bool isWebSocketRequest)
         {
             var manager = CreateConnectionManager();
             var state = manager.CreateConnection();
 
             var dispatcher = new HttpConnectionDispatcher(manager, new LoggerFactory());
 
-            var context1 = MakeRequest<TestEndPoint>("/sse", state);
+            var context1 = MakeRequest<TestEndPoint>(path, state, isWebSocketRequest: isWebSocketRequest);
-            var context2 = MakeRequest<TestEndPoint>("/sse", state);
+            var context2 = MakeRequest<TestEndPoint>(path, state, isWebSocketRequest: isWebSocketRequest);
 
             var request1 = dispatcher.ExecuteAsync<TestEndPoint>("", context1);
 
@@ -318,6 +321,34 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             Assert.False(exists);
         }
 
+        [Fact]
+        public async Task AttemptingToPollWhileAlreadyPollingReplacesTheCurrentPoll()
+        {
+            var manager = CreateConnectionManager();
+            var state = manager.CreateConnection();
+
+            var dispatcher = new HttpConnectionDispatcher(manager, new LoggerFactory());
+
+            var context1 = MakeRequest<BlockingEndPoint>("/poll", state);
+            var task1 = dispatcher.ExecuteAsync<BlockingEndPoint>("", context1);
+            var context2 = MakeRequest<BlockingEndPoint>("/poll", state);
+            var task2 = dispatcher.ExecuteAsync<BlockingEndPoint>("", context2);
+
+            // Task 1 should finish when request 2 arrives
+            await task1.OrTimeout();
+
+            // Send a message from the app to complete Task 2
+            await state.Connection.Transport.Output.WriteAsync(new Message(Encoding.UTF8.GetBytes("Hello, World"), MessageType.Text));
+
+            await task2.OrTimeout();
+
+            // Verify the results
+            Assert.Equal(StatusCodes.Status204NoContent, context1.Response.StatusCode);
+            Assert.Equal(string.Empty, GetContentAsString(context1.Response.Body));
+            Assert.Equal(StatusCodes.Status200OK, context2.Response.StatusCode);
+            Assert.Equal("T12:T:Hello, World;", GetContentAsString(context2.Response.Body));
+        }
+
         [Theory]
         [InlineData("", "text", "Hello, World", "Hello, World", MessageType.Text)] // Legacy format
         [InlineData("", "binary", "Hello, World", "Hello, World", MessageType.Binary)] // Legacy format
@@ -384,7 +415,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             return messages;
         }
 
-        private static DefaultHttpContext MakeRequest<TEndPoint>(string path, ConnectionState state, string format = null) where TEndPoint : EndPoint
+        private static DefaultHttpContext MakeRequest<TEndPoint>(string path, ConnectionState state, string format = null, bool isWebSocketRequest = false) where TEndPoint : EndPoint
         {
             var context = new DefaultHttpContext();
             var services = new ServiceCollection();
@@ -399,6 +430,14 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             }
             var qs = new QueryCollection(values);
             context.Request.Query = qs;
+            context.Response.Body = new MemoryStream();
+
+            if (isWebSocketRequest)
+            {
+                // Add Test WebSocket feature
+                context.Features.Set<IHttpWebSocketConnectionFeature>(new TestWebSocketConnectionFeature());
+            }
+
             return context;
         }
 
@@ -406,6 +445,16 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         {
             return new ConnectionManager(new Logger<ConnectionManager>(new LoggerFactory()));
         }
+
+        private string GetContentAsString(Stream body)
+        {
+            Assert.True(body.CanSeek, "Can't get content of a non-seekable stream");
+            body.Seek(0, SeekOrigin.Begin);
+            using (var reader = new StreamReader(body))
+            {
+                return reader.ReadToEnd();
+            }
+        }
     }
 
     public class BlockingEndPoint : EndPoint

test/Microsoft.AspNetCore.Sockets.Tests/Microsoft.AspNetCore.Sockets.Tests.csproj

@@ -21,6 +21,7 @@
     <PackageReference Include="Microsoft.AspNetCore.Http" Version="$(AspNetCoreVersion)" />
     <PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="$(AspNetCoreVersion)" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="$(TestSdkVersion)" />
+    <PackageReference Include="System.IO.Pipelines.Extensions" Version="$(CoreFxLabsVersion)" />
     <PackageReference Include="xunit.runner.visualstudio" Version="$(XunitVersion)" />
     <PackageReference Include="xunit" Version="$(XunitVersion)" />
   </ItemGroup>

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

@@ -0,0 +1,36 @@
+using System;
+using System.Buffers.Pools;
+using System.IO.Pipelines;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.WebSockets.Internal;
+using Microsoft.Extensions.WebSockets.Internal;
+
+namespace Microsoft.AspNetCore.Sockets.Tests
+{
+    internal class TestWebSocketConnectionFeature : IHttpWebSocketConnectionFeature, IDisposable
+    {
+        private PipeFactory _factory = new PipeFactory(ManagedBufferPool.Shared);
+
+        public bool IsWebSocketRequest => true;
+
+        public WebSocketConnection Client { get; private set; }
+
+        public ValueTask<IWebSocketConnection> AcceptWebSocketConnectionAsync(WebSocketAcceptContext context)
+        {
+            var clientToServer = _factory.Create();
+            var serverToClient = _factory.Create();
+
+            var clientSocket = new WebSocketConnection(serverToClient.Reader, clientToServer.Writer);
+            var serverSocket = new WebSocketConnection(clientToServer.Reader, serverToClient.Writer);
+
+            Client = clientSocket;
+            return new ValueTask<IWebSocketConnection>(serverSocket);
+        }
+
+        public void Dispose()
+        {
+            _factory.Dispose();
+        }
+    }
+}