diff --git a/samples/SocketsSample/project.json b/samples/SocketsSample/project.json index 400842963e..cc582adfa6 100644 --- a/samples/SocketsSample/project.json +++ b/samples/SocketsSample/project.json @@ -10,6 +10,7 @@ "Microsoft.AspNetCore.Diagnostics": "1.1.0-*", "Microsoft.AspNetCore.StaticFiles": "1.1.0-*", "Microsoft.AspNetCore.Server.IISIntegration": "1.1.0-*", + "Microsoft.AspNetCore.Server.WebListener": "0.1.0", "Microsoft.AspNetCore.Server.Kestrel": "1.1.0-*", "Microsoft.Extensions.Logging.Console": "1.1.0-*" }, diff --git a/src/Microsoft.AspNetCore.Sockets/Connection.cs b/src/Microsoft.AspNetCore.Sockets/Connection.cs index 544e1cbef3..f46b051953 100644 --- a/src/Microsoft.AspNetCore.Sockets/Connection.cs +++ b/src/Microsoft.AspNetCore.Sockets/Connection.cs @@ -10,5 +10,6 @@ namespace Microsoft.AspNetCore.Sockets { public string ConnectionId { get; set; } public IChannel Channel { get; set; } + public IDictionary Metadata { get; } = new Dictionary(); } } diff --git a/src/Microsoft.AspNetCore.Sockets/HttpConnectionDispatcher.cs b/src/Microsoft.AspNetCore.Sockets/HttpConnectionDispatcher.cs index 242db513f6..8c6cfa8e8d 100644 --- a/src/Microsoft.AspNetCore.Sockets/HttpConnectionDispatcher.cs +++ b/src/Microsoft.AspNetCore.Sockets/HttpConnectionDispatcher.cs @@ -52,8 +52,7 @@ namespace Microsoft.AspNetCore.Sockets { // Get the connection state for the current http context var connectionState = GetOrCreateConnection(context); - var channel = (HttpChannel)connectionState.Connection.Channel; - var sse = new ServerSentEvents(channel); + var sse = new ServerSentEvents(connectionState.Connection); // Register this transport for disconnect RegisterDisconnect(context, sse); @@ -76,8 +75,7 @@ namespace Microsoft.AspNetCore.Sockets { // Get the connection state for the current http context var connectionState = GetOrCreateConnection(context); - var channel = (HttpChannel)connectionState.Connection.Channel; - var ws = new WebSockets(channel); + var ws = new WebSockets(connectionState.Connection); // Register this transport for disconnect RegisterDisconnect(context, ws); @@ -128,8 +126,7 @@ namespace Microsoft.AspNetCore.Sockets var ignore = endpoint.OnConnected(connectionState.Connection); } - var channel = (HttpChannel)connectionState.Connection.Channel; - var longPolling = new LongPolling(channel); + var longPolling = new LongPolling(connectionState.Connection); // Register this transport for disconnect RegisterDisconnect(context, longPolling); @@ -144,7 +141,7 @@ namespace Microsoft.AspNetCore.Sockets private static void RegisterDisconnect(HttpContext context, IHttpTransport transport) { - context.RequestAborted.Register(state => ((IHttpTransport)state).Abort(), transport); + context.RequestAborted.Register(state => ((IHttpTransport)state).CloseAsync(), transport); } private Task ProcessGetId(HttpContext context) diff --git a/src/Microsoft.AspNetCore.Sockets/IHttpTransport.cs b/src/Microsoft.AspNetCore.Sockets/IHttpTransport.cs index dcc7e79765..b8aa396fc9 100644 --- a/src/Microsoft.AspNetCore.Sockets/IHttpTransport.cs +++ b/src/Microsoft.AspNetCore.Sockets/IHttpTransport.cs @@ -8,7 +8,17 @@ namespace Microsoft.AspNetCore.Sockets { public interface IHttpTransport { + /// + /// Executes the transport + /// + /// + /// A that completes when the transport has finished processing Task ProcessRequest(HttpContext context); - void Abort(); + + /// + /// Completes the Task returned from ProcessRequest if not already complete + /// + /// + Task CloseAsync(); } } diff --git a/src/Microsoft.AspNetCore.Sockets/LongPolling.cs b/src/Microsoft.AspNetCore.Sockets/LongPolling.cs index 6aa481833c..bbcc701251 100644 --- a/src/Microsoft.AspNetCore.Sockets/LongPolling.cs +++ b/src/Microsoft.AspNetCore.Sockets/LongPolling.cs @@ -10,14 +10,16 @@ namespace Microsoft.AspNetCore.Sockets private readonly TaskCompletionSource _initTcs = new TaskCompletionSource(); private readonly TaskCompletionSource _lifetime = new TaskCompletionSource(); private readonly HttpChannel _channel; + private readonly Connection _connection; private readonly TaskQueue _queue; private HttpContext _context; - public LongPolling(HttpChannel channel) + public LongPolling(Connection connection) { _queue = new TaskQueue(_initTcs.Task); - _channel = channel; + _connection = connection; + _channel = (HttpChannel)connection.Channel; } public async Task ProcessRequest(HttpContext context) @@ -38,7 +40,7 @@ namespace Microsoft.AspNetCore.Sockets if (buffer.IsEmpty && _channel.Output.Reading.IsCompleted) { - Abort(); + await CloseAsync(); return; } @@ -51,11 +53,24 @@ namespace Microsoft.AspNetCore.Sockets _channel.Output.Advance(buffer.End); } - - Abort(); + await EndRequest(); } - public async void Abort() + public async Task CloseAsync() + { + await _queue.Enqueue(state => + { + var context = (HttpContext)state; + // REVIEW: What happens if header was already? + context.Response.Headers["X-ASPNET-SOCKET-DISCONNECT"] = "1"; + return Task.CompletedTask; + }, + _context); + + await EndRequest(); + } + + private async Task EndRequest() { // Drain the queue and don't let any new work enter await _queue.Drain(); @@ -66,6 +81,7 @@ namespace Microsoft.AspNetCore.Sockets private Task Send(ReadableBuffer value) { + // REVIEW: Can we avoid the closure here? return _queue.Enqueue(state => { var data = (ReadableBuffer)state; diff --git a/src/Microsoft.AspNetCore.Sockets/ServerSentEvents.cs b/src/Microsoft.AspNetCore.Sockets/ServerSentEvents.cs index 2da8f5f34b..8f83eee91c 100644 --- a/src/Microsoft.AspNetCore.Sockets/ServerSentEvents.cs +++ b/src/Microsoft.AspNetCore.Sockets/ServerSentEvents.cs @@ -7,17 +7,19 @@ namespace Microsoft.AspNetCore.Sockets { public class ServerSentEvents : IHttpTransport { - private readonly TaskQueue _queue; private readonly TaskCompletionSource _initTcs = new TaskCompletionSource(); private readonly TaskCompletionSource _lifetime = new TaskCompletionSource(); private readonly HttpChannel _channel; + private readonly Connection _connection; + private readonly TaskQueue _queue; private HttpContext _context; - public ServerSentEvents(HttpChannel channel) + public ServerSentEvents(Connection connection) { _queue = new TaskQueue(_initTcs.Task); - _channel = channel; + _connection = connection; + _channel = (HttpChannel)connection.Channel; var ignore = StartSending(); } @@ -34,7 +36,7 @@ namespace Microsoft.AspNetCore.Sockets await _lifetime.Task; } - public async void Abort() + public async Task CloseAsync() { // Drain the queue so no new work can enter await _queue.Drain(); @@ -66,7 +68,7 @@ namespace Microsoft.AspNetCore.Sockets private Task Send(ReadableBuffer value) { - return _queue.Enqueue(state => + return _queue.Enqueue(async state => { var data = (ReadableBuffer)state; // TODO: Pooled buffers @@ -83,7 +85,8 @@ namespace Microsoft.AspNetCore.Sockets at += data.Length; buffer[at++] = (byte)'\n'; buffer[at++] = (byte)'\n'; - return _context.Response.Body.WriteAsync(buffer, 0, at); + await _context.Response.Body.WriteAsync(buffer, 0, at); + await _context.Response.Body.FlushAsync(); }, value); } diff --git a/src/Microsoft.AspNetCore.Sockets/WebSockets.cs b/src/Microsoft.AspNetCore.Sockets/WebSockets.cs index cbba8774c5..79e1c7c2ed 100644 --- a/src/Microsoft.AspNetCore.Sockets/WebSockets.cs +++ b/src/Microsoft.AspNetCore.Sockets/WebSockets.cs @@ -10,12 +10,14 @@ namespace Microsoft.AspNetCore.Sockets public class WebSockets : IHttpTransport { private readonly HttpChannel _channel; + private readonly Connection _connection; private readonly TaskCompletionSource _tcs = new TaskCompletionSource(); private WebSocket _ws; - public WebSockets(HttpChannel channel) + public WebSockets(Connection connection) { - _channel = channel; + _connection = connection; + _channel = (HttpChannel)connection.Channel; var ignore = StartSending(); } @@ -34,7 +36,8 @@ namespace Microsoft.AspNetCore.Sockets _tcs.TrySetResult(null); var buffer = new byte[2048]; - while (true) + + while (!_channel.Input.Writing.IsCompleted) { var result = await ws.ReceiveAsync(new ArraySegment(buffer), CancellationToken.None); @@ -49,17 +52,18 @@ namespace Microsoft.AspNetCore.Sockets } else if (result.MessageType == WebSocketMessageType.Close) { - await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None); + await ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None); // TODO: needs to remove itself from connection mamanger? break; } } } - public async void Abort() + public async Task CloseAsync() { await _tcs.Task; + // REVIEW: Close output vs Close? await _ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None); }