huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/release/2.1' into dev

This commit is contained in:
BrennanConroy 2018-03-16 16:49:59 -07:00
parent 7f54aee51c db0dc0f960
commit a43a1601c4
21 changed files with 268 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/BroadcastBenchmark.cs
View File

@ -7,6 +7,7 @@ using BenchmarkDotNet.Attributes;
using Microsoft.AspNetCore.SignalR.Internal;
using Microsoft.AspNetCore.SignalR.Internal.Protocol;
using Microsoft.AspNetCore.Sockets;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
@ -25,7 +26,7 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
[GlobalSetup]
public void GlobalSetup()
{
_hubLifetimeManager = new DefaultHubLifetimeManager<Hub>();
_hubLifetimeManager = new DefaultHubLifetimeManager<Hub>(NullLogger<DefaultHubLifetimeManager<Hub>>.Instance);
IHubProtocol protocol;

7
benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/DefaultHubDispatcherBenchmark.cs
View File

@ -36,12 +36,11 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
_dispatcher = new DefaultHubDispatcher<TestHub>(
serviceScopeFactory,
new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>()),
new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>(NullLogger<DefaultHubLifetimeManager<TestHub>>.Instance)),
new Logger<DefaultHubDispatcher<TestHub>>(NullLoggerFactory.Instance));
var options = new PipeOptions();
var pair = DuplexPipe.CreateConnectionPair(options, options);
var connection = new Sockets.DefaultConnectionContext(Guid.NewGuid().ToString(), pair.Transport, pair.Application);
var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
var connection = new Sockets.DefaultConnectionContext(Guid.NewGuid().ToString(), pair.Application, pair.Transport);
_connectionContext = new NoErrorHubConnectionContext(connection, TimeSpan.Zero, NullLoggerFactory.Instance);

2
src/Common/PipeWriterStream.cs
View File

@ -67,7 +67,7 @@ namespace System.IO.Pipelines
{
_pipeWriter.Write(source.Span);
_length += source.Length;
return new ValueTask(Task.CompletedTask);
return default;
}
#endif
}

23
src/Common/StreamExtensions.cs
View File

@ -11,10 +11,27 @@ namespace System.IO
{
internal static class StreamExtensions
{
public static async Task WriteAsync(this Stream stream, ReadOnlySequence<byte> buffer, CancellationToken cancellationToken = default)
public static ValueTask WriteAsync(this Stream stream, ReadOnlySequence<byte> buffer, CancellationToken cancellationToken = default)
{
// REVIEW: Should we special case IsSingleSegment here?
foreach (var segment in buffer)
if (buffer.IsSingleSegment)
{
#if NETCOREAPP2_1
return stream.WriteAsync(buffer.First, cancellationToken);
#else
var isArray = MemoryMarshal.TryGetArray(buffer.First, out var arraySegment);
// We're using the managed memory pool which is backed by managed buffers
Debug.Assert(isArray);
return new ValueTask(stream.WriteAsync(arraySegment.Array, arraySegment.Offset, arraySegment.Count, cancellationToken));
#endif
}
return WriteMultiSegmentAsync(stream, buffer, cancellationToken);
}
private static async ValueTask WriteMultiSegmentAsync(Stream stream, ReadOnlySequence<byte> buffer, CancellationToken cancellationToken)
{
var position = buffer.Start;
while (buffer.TryGet(ref position, out var segment))
{
#if NETCOREAPP2_1
await stream.WriteAsync(segment, cancellationToken);

36
src/Common/WebSocketExtensions.cs
View File

@ -1,12 +1,9 @@
// 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.Diagnostics;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
@ -14,29 +11,50 @@ namespace System.Net.WebSockets
{
internal static class WebSocketExtensions
{
public static Task SendAsync(this WebSocket webSocket, ReadOnlySequence<byte> buffer, WebSocketMessageType webSocketMessageType, CancellationToken cancellationToken = default)
public static ValueTask 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
if (buffer.IsSingleSegment)
{
return webSocket.SendAsync(buffer.First, webSocketMessageType, endOfMessage: true, cancellationToken).AsTask();
return webSocket.SendAsync(buffer.First, webSocketMessageType, endOfMessage: true, cancellationToken);
}
else
{
return webSocket.SendAsync(buffer.ToArray(), webSocketMessageType, endOfMessage: true, cancellationToken);
return SendMultiSegmentAsync(webSocket, buffer, webSocketMessageType, cancellationToken);
}
#else
if (buffer.IsSingleSegment)
{
var isArray = MemoryMarshal.TryGetArray(buffer.First, out var segment);
Debug.Assert(isArray);
return webSocket.SendAsync(segment, webSocketMessageType, endOfMessage: true, cancellationToken);
return new ValueTask(webSocket.SendAsync(segment, webSocketMessageType, endOfMessage: true, cancellationToken));
}
else
{
return webSocket.SendAsync(new ArraySegment<byte>(buffer.ToArray()), webSocketMessageType, true, cancellationToken);
return SendMultiSegmentAsync(webSocket, buffer, webSocketMessageType, cancellationToken);
}
#endif
}
private static async ValueTask SendMultiSegmentAsync(WebSocket webSocket, ReadOnlySequence<byte> buffer, WebSocketMessageType webSocketMessageType, CancellationToken cancellationToken = default)
{
var position = buffer.Start;
while (buffer.TryGet(ref position, out var segment))
{
#if NETCOREAPP2_1
await webSocket.SendAsync(segment, webSocketMessageType, endOfMessage: false, cancellationToken);
#else
var isArray = MemoryMarshal.TryGetArray(segment, out var arraySegment);
Debug.Assert(isArray);
await webSocket.SendAsync(arraySegment, webSocketMessageType, endOfMessage: false, cancellationToken);
#endif
}
// Empty end of message frame
#if NETCOREAPP2_1
await webSocket.SendAsync(Memory<byte>.Empty, webSocketMessageType, endOfMessage: true, cancellationToken);
#else
await webSocket.SendAsync(new ArraySegment<byte>(Array.Empty<byte>()), webSocketMessageType, endOfMessage: true, cancellationToken);
#endif
}
}

42
src/Microsoft.AspNetCore.SignalR.Core/DefaultHubLifetimeManager.cs
View File

@ -6,6 +6,7 @@ using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR.Internal.Protocol;
using Microsoft.Extensions.Logging;
namespace Microsoft.AspNetCore.SignalR
{
@ -13,6 +14,12 @@ namespace Microsoft.AspNetCore.SignalR
{
private readonly HubConnectionList _connections = new HubConnectionList();
private readonly HubGroupList _groups = new HubGroupList();
private readonly ILogger _logger;
public DefaultHubLifetimeManager(ILogger<DefaultHubLifetimeManager<THub>> logger)
{
_logger = logger;
}
public override Task AddGroupAsync(string connectionId, string groupName)
{
@ -83,7 +90,7 @@ namespace Microsoft.AspNetCore.SignalR
continue;
}
tasks.Add(connection.WriteAsync(message));
tasks.Add(SafeWriteAsync(connection, message));
}
return Task.WhenAll(tasks);
@ -105,7 +112,7 @@ namespace Microsoft.AspNetCore.SignalR
var message = CreateInvocationMessage(methodName, args);
return connection.WriteAsync(message);
return SafeWriteAsync(connection, message);
}
public override Task SendGroupAsync(string groupName, string methodName, object[] args)
@ -119,7 +126,7 @@ namespace Microsoft.AspNetCore.SignalR
if (group != null)
{
var message = CreateInvocationMessage(methodName, args);
var tasks = group.Values.Select(c => c.WriteAsync(message));
var tasks = group.Values.Select(c => SafeWriteAsync(c, message));
return Task.WhenAll(tasks);
}
@ -142,7 +149,7 @@ namespace Microsoft.AspNetCore.SignalR
var group = _groups[groupName];
if (group != null)
{
tasks.Add(Task.WhenAll(group.Values.Select(c => c.WriteAsync(message))));
tasks.Add(Task.WhenAll(group.Values.Select(c => SafeWriteAsync(c, message))));
}
}
@ -161,7 +168,7 @@ namespace Microsoft.AspNetCore.SignalR
{
var message = CreateInvocationMessage(methodName, args);
var tasks = group.Values.Where(connection => !excludedIds.Contains(connection.ConnectionId))
.Select(c => c.WriteAsync(message));
.Select(c => SafeWriteAsync(c, message));
return Task.WhenAll(tasks);
}
@ -215,5 +222,30 @@ namespace Microsoft.AspNetCore.SignalR
return userIds.Contains(connection.UserIdentifier);
});
}
// This method is to protect against connections throwing synchronously when writing to them and preventing other connections from being written to
private async Task SafeWriteAsync(HubConnectionContext connection, InvocationMessage message)
{
try
{
await connection.WriteAsync(message);
}
// This exception isn't interesting to users
catch (Exception ex)
{
Log.FailedWritingMessage(_logger, ex);
}
}
private static class Log
{
private static readonly Action<ILogger, Exception> _failedWritingMessage =
LoggerMessage.Define(LogLevel.Warning, new EventId(1, "FailedWritingMessage"), "Failed writing message.");
public static void FailedWritingMessage(ILogger logger, Exception exception)
{
_failedWritingMessage(logger, exception);
}
}
}
}

22
src/Microsoft.AspNetCore.SignalR.Redis/Internal/RedisLoggerExtensions.cs
View File

@ -12,37 +12,37 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Internal
{
// Category: RedisHubLifetimeManager<THub>
private static readonly Action<ILogger, string, Exception> _connectingToEndpoints =
LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, nameof(ConnectingToEndpoints)), "Connecting to Redis endpoints: {Endpoints}.");
LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "ConnectingToEndpoints"), "Connecting to Redis endpoints: {Endpoints}.");
private static readonly Action<ILogger, Exception> _connected =
LoggerMessage.Define(LogLevel.Information, new EventId(2, nameof(Connected)), "Connected to Redis.");
LoggerMessage.Define(LogLevel.Information, new EventId(2, "Connected"), "Connected to Redis.");
private static readonly Action<ILogger, string, Exception> _subscribing =
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(3, nameof(Subscribing)), "Subscribing to channel: {Channel}.");
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(3, "Subscribing"), "Subscribing to channel: {Channel}.");
private static readonly Action<ILogger, string, Exception> _receivedFromChannel =
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(4, nameof(ReceivedFromChannel)), "Received message from Redis channel {Channel}.");
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(4, "ReceivedFromChannel"), "Received message from Redis channel {Channel}.");
private static readonly Action<ILogger, string, Exception> _publishToChannel =
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(5, nameof(PublishToChannel)), "Publishing message to Redis channel {Channel}.");
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(5, "PublishToChannel"), "Publishing message to Redis channel {Channel}.");
private static readonly Action<ILogger, string, Exception> _unsubscribe =
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(6, nameof(Unsubscribe)), "Unsubscribing from channel: {Channel}.");
LoggerMessage.Define<string>(LogLevel.Trace, new EventId(6, "Unsubscribe"), "Unsubscribing from channel: {Channel}.");
private static readonly Action<ILogger, Exception> _notConnected =
LoggerMessage.Define(LogLevel.Warning, new EventId(7, nameof(Connected)), "Not connected to Redis.");
LoggerMessage.Define(LogLevel.Warning, new EventId(7, "Connected"), "Not connected to Redis.");
private static readonly Action<ILogger, Exception> _connectionRestored =
LoggerMessage.Define(LogLevel.Information, new EventId(8, nameof(ConnectionRestored)), "Connection to Redis restored.");
LoggerMessage.Define(LogLevel.Information, new EventId(8, "ConnectionRestored"), "Connection to Redis restored.");
private static readonly Action<ILogger, Exception> _connectionFailed =
LoggerMessage.Define(LogLevel.Warning, new EventId(9, nameof(ConnectionFailed)), "Connection to Redis failed.");
LoggerMessage.Define(LogLevel.Warning, new EventId(9, "ConnectionFailed"), "Connection to Redis failed.");
private static readonly Action<ILogger, Exception> _failedWritingMessage =
LoggerMessage.Define(LogLevel.Warning, new EventId(10, nameof(FailedWritingMessage)), "Failed writing message.");
LoggerMessage.Define(LogLevel.Warning, new EventId(10, "FailedWritingMessage"), "Failed writing message.");
private static readonly Action<ILogger, Exception> _internalMessageFailed =
LoggerMessage.Define(LogLevel.Warning, new EventId(11, nameof(InternalMessageFailed)), "Error processing message for internal server message.");
LoggerMessage.Define(LogLevel.Warning, new EventId(11, "InternalMessageFailed"), "Error processing message for internal server message.");
public static void ConnectingToEndpoints(this ILogger logger, EndPointCollection endpoints)
{

66
src/Microsoft.AspNetCore.SignalR.Redis/RedisHubLifetimeManager.cs
View File

@ -173,7 +173,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
var connection = _connections[connectionId];
if (connection != null)
{
return connection.WriteAsync(message.CreateInvocation());
return SafeWriteAsync(connection, message.CreateInvocation());
}
return PublishAsync(_channelNamePrefix + "." + connectionId, message);
@ -402,14 +402,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
var invocation = message.CreateInvocation();
foreach (var connection in _connections)
{
try
{
tasks.Add(connection.WriteAsync(invocation));
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
tasks.Add(SafeWriteAsync(connection, invocation));
}
await Task.WhenAll(tasks);
@ -441,14 +434,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
{
if (!excludedIds.Contains(connection.ConnectionId))
{
try
{
tasks.Add(connection.WriteAsync(invocation));
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
tasks.Add(SafeWriteAsync(connection, invocation));
}
}
@ -524,16 +510,9 @@ namespace Microsoft.AspNetCore.SignalR.Redis
_logger.Subscribing(connectionChannel);
return _bus.SubscribeAsync(connectionChannel, async (c, data) =>
{
try
{
var message = DeserializeMessage<RedisInvocationMessage>(data);
var message = DeserializeMessage<RedisInvocationMessage>(data);
await connection.WriteAsync(message.CreateInvocation());
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
await SafeWriteAsync(connection, message.CreateInvocation());
});
}
@ -545,16 +524,9 @@ namespace Microsoft.AspNetCore.SignalR.Redis
// TODO: Look at optimizing (looping over connections checking for Name)
return _bus.SubscribeAsync(userChannel, async (c, data) =>
{
try
{
var message = DeserializeMessage<RedisInvocationMessage>(data);
var message = DeserializeMessage<RedisInvocationMessage>(data);
await connection.WriteAsync(message.CreateInvocation());
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
await SafeWriteAsync(connection, message.CreateInvocation());
});
}
@ -576,14 +548,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
continue;
}
try
{
tasks.Add(groupConnection.WriteAsync(invocation));
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
tasks.Add(SafeWriteAsync(groupConnection, invocation));
}
await Task.WhenAll(tasks);
@ -611,7 +576,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
// This also saves serializing and deserializing the message!
if (connection != null)
{
publishTasks.Add(connection.WriteAsync(message.CreateInvocation()));
publishTasks.Add(SafeWriteAsync(connection, message.CreateInvocation()));
}
else
{
@ -662,6 +627,19 @@ namespace Microsoft.AspNetCore.SignalR.Redis
return Task.CompletedTask;
}
// This method is to protect against connections throwing synchronously when writing to them and preventing other connections from being written to
private async Task SafeWriteAsync(HubConnectionContext connection, InvocationMessage message)
{
try
{
await connection.WriteAsync(message);
}
catch (Exception ex)
{
_logger.FailedWritingMessage(ex);
}
}
private class LoggerTextWriter : TextWriter
{
private readonly ILogger _logger;

32
src/Microsoft.AspNetCore.Sockets.Abstractions/DefaultConnectionContext.cs
View File

@ -22,16 +22,20 @@ namespace Microsoft.AspNetCore.Sockets
IConnectionHeartbeatFeature,
ITransferFormatFeature
{
private List<(Action<object> handler, object state)> _heartbeatHandlers = new List<(Action<object> handler, object state)>();
private object _heartbeatLock = new object();
private List<(Action<object> handler, object state)> _heartbeatHandlers;
// This tcs exists so that multiple calls to DisposeAsync all wait asynchronously
// on the same task
private TaskCompletionSource<object> _disposeTcs = new TaskCompletionSource<object>();
public DefaultConnectionContext(string id, IDuplexPipe transport, IDuplexPipe application)
/// <summary>
/// Creates the DefaultConnectionContext without Pipes to avoid upfront allocations.
/// The caller is expected to set the <see cref="Transport"/> and <see cref="Application"/> pipes manually.
/// </summary>
/// <param name="id"></param>
public DefaultConnectionContext(string id)
{
Transport = transport;
Application = application;
ConnectionId = id;
LastSeenUtc = DateTime.UtcNow;
@ -50,6 +54,13 @@ namespace Microsoft.AspNetCore.Sockets
Features.Set<ITransferFormatFeature>(this);
}
public DefaultConnectionContext(string id, IDuplexPipe transport, IDuplexPipe application)
: this(id)
{
Transport = transport;
Application = application;
}
public CancellationTokenSource Cancellation { get; set; }
public SemaphoreSlim Lock { get; } = new SemaphoreSlim(1, 1);
@ -80,16 +91,25 @@ namespace Microsoft.AspNetCore.Sockets
public void OnHeartbeat(Action<object> action, object state)
{
lock (_heartbeatHandlers)
lock (_heartbeatLock)
{
if (_heartbeatHandlers == null)
{
_heartbeatHandlers = new List<(Action<object> handler, object state)>();
}
_heartbeatHandlers.Add((action, state));
}
}
public void TickHeartbeat()
{
lock (_heartbeatHandlers)
lock (_heartbeatLock)
{
if (_heartbeatHandlers == null)
{
return;
}
foreach (var (handler, state) in _heartbeatHandlers)
{
handler(state);

17
src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/StreamExtensions.cs
View File

@ -12,7 +12,9 @@ namespace System.IO.Pipelines
{
try
{
await stream.CopyToAsync(writer, cancellationToken);
// REVIEW: Should we use the default buffer size here?
// 81920 is the default bufferSize, there is no stream.CopyToAsync overload that takes only a cancellationToken
await stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken);
}
catch (Exception ex)
{
@ -22,19 +24,6 @@ namespace System.IO.Pipelines
writer.Complete();
}
/// <summary>
/// Copies the content of a <see cref="Stream"/> into a <see cref="PipeWriter"/>.
/// </summary>
/// <param name="stream"></param>
/// <param name="writer"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private static Task CopyToAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default)
{
// 81920 is the default bufferSize, there is not stream.CopyToAsync overload that takes only a cancellationToken
return stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken);
}
private class PipelineWriterStream : Stream
{
private readonly PipeWriter _writer;

2
src/Microsoft.AspNetCore.Sockets.Client.Http/SendUtils.cs
View File

@ -111,7 +111,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
protected override Task SerializeToStreamAsync(Stream stream, TransportContext context)
{
return stream.WriteAsync(_buffer);
return stream.WriteAsync(_buffer).AsTask();
}
protected override bool TryComputeLength(out long length)

30
src/Microsoft.AspNetCore.Sockets.Http/ConnectionManager.cs
View File

@ -2,12 +2,14 @@
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
using System.Buffers.Text;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.IO.Pipelines;
using System.Net.WebSockets;
using System.Security.Cryptography;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Hosting;
@ -22,6 +24,8 @@ namespace Microsoft.AspNetCore.Sockets
// TODO: Consider making this configurable? At least for testing?
private static readonly TimeSpan _heartbeatTickRate = TimeSpan.FromSeconds(1);
private static readonly RNGCryptoServiceProvider _keyGenerator = new RNGCryptoServiceProvider();
private readonly ConcurrentDictionary<string, (DefaultConnectionContext Connection, ValueStopwatch Timer)> _connections = new ConcurrentDictionary<string, (DefaultConnectionContext Connection, ValueStopwatch Timer)>();
private Timer _timer;
private readonly ILogger<ConnectionManager> _logger;
@ -63,23 +67,31 @@ namespace Microsoft.AspNetCore.Sockets
return false;
}
public DefaultConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
/// <summary>
/// Creates a connection without Pipes setup to allow saving allocations until Pipes are needed.
/// </summary>
/// <returns></returns>
public DefaultConnectionContext CreateConnection()
{
var id = MakeNewConnectionId();
_logger.CreatedNewConnection(id);
var connectionTimer = SocketEventSource.Log.ConnectionStart(id);
var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
var connection = new DefaultConnectionContext(id, pair.Application, pair.Transport);
var connection = new DefaultConnectionContext(id);
_connections.TryAdd(id, (connection, connectionTimer));
return connection;
}
public DefaultConnectionContext CreateConnection()
public DefaultConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
{
return CreateConnection(PipeOptions.Default, PipeOptions.Default);
var connection = CreateConnection();
var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
connection.Application = pair.Transport;
connection.Transport = pair.Application;
return connection;
}
public void RemoveConnection(string id)
@ -94,8 +106,12 @@ namespace Microsoft.AspNetCore.Sockets
private static string MakeNewConnectionId()
{
// TODO: We need to sign and encyrpt this
return Guid.NewGuid().ToString();
// TODO: Use Span when WebEncoders implements Span methods https://github.com/aspnet/Home/issues/2966
// 128 bit buffer / 8 bits per byte = 16 bytes
var buffer = new byte[16];
_keyGenerator.GetBytes(buffer);
// Generate the id with RNGCrypto because we want a cryptographically random id, which GUID is not
return WebEncoders.Base64UrlEncode(buffer);
}
private static void Scan(object state)

35
src/Microsoft.AspNetCore.Sockets.Http/HttpConnectionDispatcher.cs
View File

@ -49,7 +49,7 @@ namespace Microsoft.AspNetCore.Sockets
if (HttpMethods.IsPost(context.Request.Method))
{
// POST /{path}
await ProcessSend(context);
await ProcessSend(context, options);
}
else if (HttpMethods.IsGet(context.Request.Method))
{
@ -99,7 +99,7 @@ namespace Microsoft.AspNetCore.Sockets
if (headers.Accept?.Contains(new Net.Http.Headers.MediaTypeHeaderValue("text/event-stream")) == true)
{
// Connection must already exist
var connection = await GetConnectionAsync(context);
var connection = await GetConnectionAsync(context, options);
if (connection == null)
{
// No such connection, GetConnection already set the response status code
@ -149,7 +149,7 @@ namespace Microsoft.AspNetCore.Sockets
// GET /{path} maps to long polling
// Connection must already exist
var connection = await GetConnectionAsync(context);
var connection = await GetConnectionAsync(context, options);
if (connection == null)
{
// No such connection, GetConnection already set the response status code
@ -361,7 +361,7 @@ namespace Microsoft.AspNetCore.Sockets
context.Response.ContentType = "application/json";
// Establish the connection
var connection = CreateConnectionInternal(options);
var connection = _manager.CreateConnection();
// Set the Connection ID on the logging scope so that logs from now on will have the
// Connection ID metadata set.
@ -429,9 +429,9 @@ namespace Microsoft.AspNetCore.Sockets
private static string GetConnectionId(HttpContext context) => context.Request.Query["id"];
private async Task ProcessSend(HttpContext context)
private async Task ProcessSend(HttpContext context, HttpSocketOptions options)
{
var connection = await GetConnectionAsync(context);
var connection = await GetConnectionAsync(context, options);
if (connection == null)
{
// No such connection, GetConnection already set the response status code
@ -505,7 +505,7 @@ namespace Microsoft.AspNetCore.Sockets
return true;
}
private async Task<DefaultConnectionContext> GetConnectionAsync(HttpContext context)
private async Task<DefaultConnectionContext> GetConnectionAsync(HttpContext context, HttpSocketOptions options)
{
var connectionId = GetConnectionId(context);
@ -527,16 +527,25 @@ namespace Microsoft.AspNetCore.Sockets
return null;
}
EnsureConnectionStateInternal(connection, options);
return connection;
}
private DefaultConnectionContext CreateConnectionInternal(HttpSocketOptions options)
private void EnsureConnectionStateInternal(DefaultConnectionContext connection, HttpSocketOptions options)
{
var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
return _manager.CreateConnection(transportPipeOptions, appPipeOptions);
// If the connection doesn't have a pipe yet then create one, we lazily create the pipe to save on allocations until the client actually connects
if (connection.Transport == null)
{
var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
connection.Transport = pair.Application;
connection.Application = pair.Transport;
}
}
// This is only used for WebSockets connections, which can connect directly without negotiating
private async Task<DefaultConnectionContext> GetOrCreateConnectionAsync(HttpContext context, HttpSocketOptions options)
{
var connectionId = GetConnectionId(context);
@ -545,7 +554,7 @@ namespace Microsoft.AspNetCore.Sockets
// There's no connection id so this is a brand new connection
if (StringValues.IsNullOrEmpty(connectionId))
{
connection = CreateConnectionInternal(options);
connection = _manager.CreateConnection();
}
else if (!_manager.TryGetConnection(connectionId, out connection))
{
@ -555,6 +564,8 @@ namespace Microsoft.AspNetCore.Sockets
return null;
}
EnsureConnectionStateInternal(connection, options);
return connection;
}
}

3
src/Microsoft.AspNetCore.Sockets.Http/Microsoft.AspNetCore.Sockets.Http.csproj
View File

@ -10,7 +10,7 @@
<Compile Include="..\Common\WebSocketExtensions.cs" Link="WebSocketExtensions.cs" />
<Compile Include="..\Common\StreamExtensions.cs" Link="StreamExtensions.cs" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Microsoft.AspNetCore.Sockets.Common.Http\Microsoft.AspNetCore.Sockets.Common.Http.csproj" />
<ProjectReference Include="..\Microsoft.AspNetCore.Sockets.Abstractions\Microsoft.AspNetCore.Sockets.Abstractions.csproj" />
@ -22,6 +22,7 @@
<PackageReference Include="Microsoft.AspNetCore.Routing" Version="$(MicrosoftAspNetCoreRoutingPackageVersion)" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="$(MicrosoftAspNetCoreWebSocketsPackageVersion)" />
<PackageReference Include="Microsoft.Extensions.SecurityHelper.Sources" PrivateAssets="All" Version="$(MicrosoftExtensionsSecurityHelperSourcesPackageVersion)" />
<PackageReference Include="Microsoft.Extensions.WebEncoders.Sources" Version="$(MicrosoftExtensionsWebEncodersSourcesPackageVersion)" PrivateAssets="All" />
<PackageReference Include="Microsoft.Extensions.ValueStopwatch.Sources" Version="$(MicrosoftExtensionsValueStopwatchSourcesPackageVersion)" PrivateAssets="All" />
<PackageReference Include="Newtonsoft.Json" Version="$(NewtonsoftJsonPackageVersion)" />
</ItemGroup>

5
test/Microsoft.AspNetCore.SignalR.Redis.Tests/RedisHubLifetimeManagerTests.cs
View File

@ -503,7 +503,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Tests
}
[Fact]
public async Task WritingToLocalConnectionThatFailsThrowsException()
public async Task WritingToLocalConnectionThatFailsDoesNotThrowException()
{
var manager = new RedisHubLifetimeManager<MyHub>(new LoggerFactory().CreateLogger<RedisHubLifetimeManager<MyHub>>(), Options.Create(new RedisOptions()
{
@ -519,8 +519,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Tests
await manager.OnConnectedAsync(connection).OrTimeout();
var exception = await Assert.ThrowsAsync<Exception>(() => manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout());
Assert.Equal("Message", exception.Message);
await manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout();
}
}

65
test/Microsoft.AspNetCore.SignalR.Tests/DefaultHubLifetimeManagerTests.cs
View File

@ -3,6 +3,8 @@ using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR.Internal.Protocol;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using Xunit;
@ -11,12 +13,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
public class DefaultHubLifetimeManagerTests
{
[Fact]
public async Task InvokeAllAsyncWritesToAllConnectionsOutput()
public async Task SendAllAsyncWritesToAllConnectionsOutput()
{
using (var client1 = new TestClient())
using (var client2 = new TestClient())
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connection1 = HubConnectionContextUtils.Create(client1.Connection);
var connection2 = HubConnectionContextUtils.Create(client2.Connection);
@ -38,12 +40,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
[Fact]
public async Task InvokeAllAsyncDoesNotWriteToDisconnectedConnectionsOutput()
public async Task SendAllAsyncDoesNotWriteToDisconnectedConnectionsOutput()
{
using (var client1 = new TestClient())
using (var client2 = new TestClient())
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connection1 = HubConnectionContextUtils.Create(client1.Connection);
var connection2 = HubConnectionContextUtils.Create(client2.Connection);
@ -64,12 +66,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
[Fact]
public async Task InvokeGroupAsyncWritesToAllConnectionsInGroupOutput()
public async Task SendGroupAsyncWritesToAllConnectionsInGroupOutput()
{
using (var client1 = new TestClient())
using (var client2 = new TestClient())
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connection1 = HubConnectionContextUtils.Create(client1.Connection);
var connection2 = HubConnectionContextUtils.Create(client2.Connection);
@ -90,11 +92,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
[Fact]
public async Task InvokeConnectionAsyncWritesToConnectionOutput()
public async Task SendConnectionAsyncWritesToConnectionOutput()
{
using (var client = new TestClient())
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connection = HubConnectionContextUtils.Create(client.Connection);
await manager.OnConnectedAsync(connection).OrTimeout();
@ -109,42 +111,71 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
[Fact]
public async Task InvokeConnectionAsyncThrowsIfConnectionFailsToWrite()
public async Task SendConnectionAsyncDoesNotThrowIfConnectionFailsToWrite()
{
using (var client = new TestClient())
{
// Force an exception when writing to connection
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connectionMock = HubConnectionContextUtils.CreateMock(client.Connection);
// Force an exception when writing to connection
connectionMock.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Throws(new Exception("Message"));
var connection = connectionMock.Object;
await manager.OnConnectedAsync(connection).OrTimeout();
var exception = await Assert.ThrowsAsync<Exception>(() => manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout());
Assert.Equal("Message", exception.Message);
await manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout();
}
}
[Fact]
public async Task InvokeConnectionAsyncOnNonExistentConnectionNoops()
public async Task SendAllAsyncSendsToAllConnectionsEvenWhenSomeFailToSend()
{
var manager = new DefaultHubLifetimeManager<MyHub>();
using (var client = new TestClient())
using (var client2 = new TestClient())
{
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
var connectionMock = HubConnectionContextUtils.CreateMock(client.Connection);
var connectionMock2 = HubConnectionContextUtils.CreateMock(client2.Connection);
var tcs = new TaskCompletionSource<object>();
var tcs2 = new TaskCompletionSource<object>();
// Force an exception when writing to connection
connectionMock.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Callback(() => tcs.TrySetResult(null)).Throws(new Exception("Message"));
connectionMock2.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Callback(() => tcs2.TrySetResult(null)).Throws(new Exception("Message"));
var connection = connectionMock.Object;
var connection2 = connectionMock2.Object;
await manager.OnConnectedAsync(connection).OrTimeout();
await manager.OnConnectedAsync(connection2).OrTimeout();
await manager.SendAllAsync("Hello", new object[] { "World" }).OrTimeout();
// Check that all connections were "written" to
await tcs.Task.OrTimeout();
await tcs2.Task.OrTimeout();
}
}
[Fact]
public async Task SendConnectionAsyncOnNonExistentConnectionNoops()
{
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
await manager.SendConnectionAsync("NotARealConnectionId", "Hello", new object[] { "World" }).OrTimeout();
}
[Fact]
public async Task AddGroupOnNonExistentConnectionNoops()
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
await manager.AddGroupAsync("NotARealConnectionId", "MyGroup").OrTimeout();
}
[Fact]
public async Task RemoveGroupOnNonExistentConnectionNoops()
{
var manager = new DefaultHubLifetimeManager<MyHub>();
var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
await manager.RemoveGroupAsync("NotARealConnectionId", "MyGroup").OrTimeout();
}

23
test/Microsoft.AspNetCore.Sockets.Tests/ConnectionManagerTests.cs
View File

@ -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.IO.Pipelines;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Logging;
@ -22,8 +23,8 @@ namespace Microsoft.AspNetCore.Sockets.Tests
Assert.Null(connection.ApplicationTask);
Assert.Null(connection.TransportTask);
Assert.Null(connection.Cancellation);
Assert.NotEqual(default(DateTime), connection.LastSeenUtc);
Assert.NotNull(connection.Transport);
Assert.NotEqual(default, connection.LastSeenUtc);
Assert.Null(connection.Transport);
}
[Fact]
@ -42,7 +43,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public void AddNewConnection()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
var transport = connection.Transport;
@ -58,7 +59,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public void RemoveConnection()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
var transport = connection.Transport;
@ -77,7 +78,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public async Task CloseConnectionsEndsAllPendingConnections()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
connection.ApplicationTask = Task.Run(async () =>
{
@ -89,7 +90,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
}
finally
{
connection.Transport.Input.AdvanceTo(result.Buffer.End);
connection.Transport.Input.AdvanceTo(result.Buffer.End);
}
});
@ -115,7 +116,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public async Task DisposingConnectionMultipleTimesWaitsOnConnectionClose()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
connection.ApplicationTask = tcs.Task;
@ -135,7 +136,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public async Task DisposingConnectionMultipleGetsExceptionFromTransportOrApp()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
connection.ApplicationTask = tcs.Task;
@ -159,7 +160,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public async Task DisposingConnectionMultipleGetsCancellation()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
connection.ApplicationTask = tcs.Task;
@ -180,7 +181,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
public async Task DisposeInactiveConnection()
{
var connectionManager = CreateConnectionManager();
var connection = connectionManager.CreateConnection();;
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
Assert.NotNull(connection.ConnectionId);
Assert.NotNull(connection.Transport);
@ -209,7 +210,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
appLifetime.Start();
var connection = connectionManager.CreateConnection();
var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
connection.Application.Output.OnReaderCompleted((error, state) =>
{

3
test/Microsoft.AspNetCore.Sockets.Tests/HttpConnectionDispatcherTests.cs
View File

@ -77,7 +77,10 @@ namespace Microsoft.AspNetCore.Sockets.Tests
await dispatcher.ExecuteNegotiateAsync(context, httpSocketOptions);
var negotiateResponse = JsonConvert.DeserializeObject<JObject>(Encoding.UTF8.GetString(ms.ToArray()));
var connectionId = negotiateResponse.Value<string>("connectionId");
context.Request.QueryString = context.Request.QueryString.Add("id", connectionId);
Assert.True(manager.TryGetConnection(connectionId, out var connection));
// Fake actual connection after negotiate to populate the pipes on the connection
await dispatcher.ExecuteAsync(context, httpSocketOptions, c => Task.CompletedTask);
// This write should complete immediately but it exceeds the writer threshold
var writeTask = connection.Application.Output.WriteAsync(new byte[] { (byte)'b', (byte)'y', (byte)'t', (byte)'e', (byte)'s' });

1
test/Microsoft.AspNetCore.Sockets.Tests/LongPollingTests.cs
View File

@ -21,6 +21,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
{
var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
var connection = new DefaultConnectionContext("foo", pair.Transport, pair.Application);
var context = new DefaultHttpContext();
var poll = new LongPollingTransport(CancellationToken.None, connection.Application.Input, connectionId: string.Empty, loggerFactory: new LoggerFactory());

2
test/Microsoft.AspNetCore.Sockets.Tests/ServerSentEventsTests.cs
View File

@ -4,7 +4,6 @@
using System.IO;
using System.IO.Pipelines;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Http.Features;
@ -58,7 +57,6 @@ namespace Microsoft.AspNetCore.Sockets.Tests
var connection = new DefaultConnectionContext("foo", pair.Transport, pair.Application);
var context = new DefaultHttpContext();
var ms = new MemoryStream();
context.Response.Body = ms;
var sse = new ServerSentEventsTransport(connection.Application.Input, connectionId: string.Empty, loggerFactory: new LoggerFactory());

1
test/Microsoft.AspNetCore.Sockets.Tests/WebSocketsTests.cs
View File

@ -320,6 +320,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
// We want to verify behavior without timeout affecting it
CloseTimeout = TimeSpan.FromSeconds(20)
};
var connectionContext = new DefaultConnectionContext(string.Empty, null, null);
var ws = new WebSocketsTransport(options, connection.Application, connectionContext, loggerFactory);