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

Complete logging for Socket.* packages (#655)

This commit is contained in:
BrennanConroy 2017-07-12 16:16:09 -07:00 committed by GitHub
parent cd5d4fbdaf
commit bcefbae00c
16 changed files with 669 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs
View File

@ -28,6 +28,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
private TaskCompletionSource<object> _startTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
private TaskQueue _eventQueue = new TaskQueue();
private readonly ITransportFactory _transportFactory;
private string _connectionId;
private ReadableChannel<byte[]> Input => _transportChannel.In;
private WritableChannel<SendMessage> Output => _transportChannel.Out;
@ -111,23 +112,24 @@ namespace Microsoft.AspNetCore.Sockets.Client
private async Task StartAsyncInternal()
{
_logger.LogDebug("Starting connection.");
_logger.HttpConnectionStarting();
try
{
var negotiationResponse = await Negotiate(Url, _httpClient, _logger);
_connectionId = negotiationResponse.ConnectionId;
// Connection is being stopped while start was in progress
if (_connectionState == ConnectionState.Disconnected)
{
_logger.LogDebug("Connection was closed from a different thread.");
_logger.HttpConnectionClosed(_connectionId);
return;
}
_transport = _transportFactory.CreateTransport(GetAvailableServerTransports(negotiationResponse));
var connectUrl = CreateConnectUrl(Url, negotiationResponse);
_logger.LogDebug("Starting transport '{0}' with Url: {1}", _transport.GetType().Name, connectUrl);
_logger.StartingTransport(_connectionId, _transport.GetType().Name, connectUrl);
await StartTransport(connectUrl);
}
catch
@ -142,7 +144,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
var ignore = _eventQueue.Enqueue(() =>
{
_logger.LogDebug("Raising Connected event");
_logger.RaiseConnected(_connectionId);
Connected?.Invoke();
@ -158,17 +160,17 @@ namespace Microsoft.AspNetCore.Sockets.Client
// to make sure that the message removed from the channel is processed before we drain the queue.
// There is a short window between we start the channel and assign the _receiveLoopTask a value.
// To make sure that _receiveLoopTask can be awaited (i.e. is not null) we need to await _startTask.
_logger.LogDebug("Ensuring all outstanding messages are processed.");
_logger.ProcessRemainingMessages(_connectionId);
await _startTcs.Task;
await _receiveLoopTask;
_logger.LogDebug("Draining event queue");
_logger.DrainEvents(_connectionId);
await _eventQueue.Drain();
_httpClient.Dispose();
_logger.LogDebug("Raising Closed event");
_logger.RaiseClosed(_connectionId);
Closed?.Invoke(t.IsFaulted ? t.Exception.InnerException : null);
@ -186,7 +188,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
try
{
// Get a connection ID from the server
logger.LogDebug("Establishing Connection at: {0}", url);
logger.EstablishingConnection(url);
using (var request = new HttpRequestMessage(HttpMethod.Options, url))
using (var response = await httpClient.SendAsync(request))
{
@ -196,7 +198,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
catch (Exception ex)
{
logger.LogError("Failed to start connection. Error getting negotiation response from '{0}': {1}", url, ex);
logger.ErrorWithNegotiation(url, ex);
throw;
}
}
@ -260,11 +262,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
// Start the transport, giving it one end of the pipeline
try
{
await _transport.StartAsync(connectUrl, applicationSide);
await _transport.StartAsync(connectUrl, applicationSide, _connectionId);
}
catch (Exception ex)
{
_logger.LogError("Failed to start connection. Error starting transport '{0}': {1}", _transport.GetType().Name, ex);
_logger.ErrorStartingTransport(_connectionId, _transport.GetType().Name, ex);
throw;
}
}
@ -273,13 +275,13 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
try
{
_logger.LogTrace("Beginning receive loop.");
_logger.HttpReceiveStarted(_connectionId);
while (await Input.WaitToReadAsync())
{
if (_connectionState != ConnectionState.Connected)
{
_logger.LogDebug("Message received but connection is not connected. Skipping raising Received event.");
_logger.SkipRaisingReceiveEvent(_connectionId);
// drain
Input.TryRead(out _);
continue;
@ -287,10 +289,10 @@ namespace Microsoft.AspNetCore.Sockets.Client
if (Input.TryRead(out var buffer))
{
_logger.LogDebug("Scheduling raising Received event.");
_logger.ScheduleReceiveEvent(_connectionId);
_ = _eventQueue.Enqueue(() =>
{
_logger.LogDebug("Raising Received event.");
_logger.RaiseReceiveEvent(_connectionId);
// Making a copy of the Received handler to ensure that its not null
// Can't use the ? operator because we specifically want to check if the handler is null
@ -305,7 +307,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
else
{
_logger.LogDebug("Could not read message.");
_logger.FailedReadingMessage(_connectionId);
}
}
@ -314,10 +316,10 @@ namespace Microsoft.AspNetCore.Sockets.Client
catch (Exception ex)
{
Output.TryComplete(ex);
_logger.LogError("Error receiving message: {0}", ex);
_logger.ErrorReceiving(_connectionId, ex);
}
_logger.LogTrace("Ending receive loop");
_logger.EndReceive(_connectionId);
}
public async Task SendAsync(byte[] data, CancellationToken cancellationToken = default(CancellationToken))
@ -340,7 +342,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
var sendTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
var message = new SendMessage(data, sendTcs);
_logger.LogDebug("Sending message");
_logger.SendingMessage(_connectionId);
while (await Output.WaitToWriteAsync(cancellationToken))
{
@ -354,7 +356,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
public async Task DisposeAsync()
{
_logger.LogInformation("Stopping client.");
_logger.StoppingClient(_connectionId);
if (Interlocked.Exchange(ref _connectionState, ConnectionState.Disconnected) == ConnectionState.Initial)
{

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

@ -9,7 +9,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
public interface ITransport
{
Task StartAsync(Uri url, Channel<byte[], SendMessage> application);
Task StartAsync(Uri url, Channel<byte[], SendMessage> application, string connectionId);
Task StopAsync();
}
}

502
src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/SocketClientLoggerExtensions.cs Normal file
View File

@ -0,0 +1,502 @@
// 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.Net.WebSockets;
using Microsoft.Extensions.Logging;
namespace Microsoft.AspNetCore.Sockets.Client.Internal
{
internal static class SocketClientLoggerExtensions
{
// Category: Shared with LongPollingTransport, WebSocketsTransport and ServerSentEventsTransport
private static readonly Action<ILogger, DateTime, string, Exception> _startTransport =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 0, "{time}: Connection Id {connectionId}: Starting transport.");
private static readonly Action<ILogger, DateTime, string, Exception> _transportStopped =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 1, "{time}: Connection Id {connectionId}: Transport stopped.");
private static readonly Action<ILogger, DateTime, string, Exception> _startReceive =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 2, "{time}: Connection Id {connectionId}: Starting receive loop.");
private static readonly Action<ILogger, DateTime, string, Exception> _receiveStopped =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 3, "{time}: Connection Id {connectionId}: Receive loop stopped.");
private static readonly Action<ILogger, DateTime, string, Exception> _receiveCanceled =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 4, "{time}: Connection Id {connectionId}: Receive loop canceled.");
private static readonly Action<ILogger, DateTime, string, Exception> _transportStopping =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 5, "{time}: Connection Id {connectionId}: Transport is stopping.");
private static readonly Action<ILogger, DateTime, string, Exception> _sendStarted =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 6, "{time}: Connection Id {connectionId}: Starting the send loop.");
private static readonly Action<ILogger, DateTime, string, Exception> _sendStopped =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 7, "{time}: Connection Id {connectionId}: Send loop stopped.");
private static readonly Action<ILogger, DateTime, string, Exception> _sendCanceled =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 8, "{time}: Connection Id {connectionId}: Send loop canceled.");
// Category: WebSocketsTransport
private static readonly Action<ILogger, DateTime, string, WebSocketCloseStatus?, Exception> _webSocketClosed =
LoggerMessage.Define<DateTime, string, WebSocketCloseStatus?>(LogLevel.Information, 9, "{time}: Connection Id {connectionId}: Websocket closed by the server. Close status {closeStatus}.");
private static readonly Action<ILogger, DateTime, string, WebSocketMessageType, int, bool, Exception> _messageReceived =
LoggerMessage.Define<DateTime, string, WebSocketMessageType, int, bool>(LogLevel.Debug, 10, "{time}: Connection Id {connectionId}: Message received. Type: {messageType}, size: {count}, EndOfMessage: {endOfMessage}.");
private static readonly Action<ILogger, DateTime, string, int, Exception> _messageToApp =
LoggerMessage.Define<DateTime, string, int>(LogLevel.Information, 11, "{time}: Connection Id {connectionId}: Passing message to application. Payload size: {count}.");
private static readonly Action<ILogger, DateTime, string, int, Exception> _receivedFromApp =
LoggerMessage.Define<DateTime, string, int>(LogLevel.Debug, 12, "{time}: Connection Id {connectionId}: Received message from application. Payload size: {count}.");
private static readonly Action<ILogger, DateTime, string, Exception> _sendMessageCanceled =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 13, "{time}: Connection Id {connectionId}: Sending a message canceled.");
private static readonly Action<ILogger, DateTime, string, Exception> _errorSendingMessage =
LoggerMessage.Define<DateTime, string>(LogLevel.Error, 14, "{time}: Connection Id {connectionId}: Error while sending a message.");
private static readonly Action<ILogger, DateTime, string, Exception> _closingWebSocket =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 15, "{time}: Connection Id {connectionId}: Closing WebSocket.");
private static readonly Action<ILogger, DateTime, string, Exception> _closingWebSocketFailed =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 16, "{time}: Connection Id {connectionId}: Closing webSocket failed.");
// Category: ServerSentEventsTransport and LongPollingTransport
private static readonly Action<ILogger, DateTime, string, int, Uri, Exception> _sendingMessages =
LoggerMessage.Define<DateTime, string, int, Uri>(LogLevel.Debug, 9, "{time}: Connection Id {connectionId}: Sending {count} message(s) to the server using url: {url}.");
private static readonly Action<ILogger, DateTime, string, Exception> _sentSuccessfully =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 10, "{time}: Connection Id {connectionId}: Message(s) sent successfully.");
private static readonly Action<ILogger, DateTime, string, Exception> _noMessages =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 11, "{time}: Connection Id {connectionId}: No messages in batch to send.");
private static readonly Action<ILogger, DateTime, string, Uri, Exception> _errorSending =
LoggerMessage.Define<DateTime, string, Uri>(LogLevel.Error, 12, "{time}: Connection Id {connectionId}: Error while sending to '{url}'.");
// Category: ServerSentEventsTransport
private static readonly Action<ILogger, DateTime, string, Exception> _eventStreamEnded =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 13, "{time}: Connection Id {connectionId}: Server-Sent Event Stream ended.");
// Category: LongPollingTransport
private static readonly Action<ILogger, DateTime, string, Exception> _closingConnection =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 13, "{time}: Connection Id {connectionId}: The server is closing the connection.");
private static readonly Action<ILogger, DateTime, string, Exception> _receivedMessages =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 14, "{time}: Connection Id {connectionId}: Received messages from the server.");
private static readonly Action<ILogger, DateTime, string, Uri, Exception> _errorPolling =
LoggerMessage.Define<DateTime, string, Uri>(LogLevel.Error, 15, "{time}: Connection Id {connectionId}: Error while polling '{pollUrl}'.");
// Category: HttpConnection
private static readonly Action<ILogger, DateTime, Exception> _httpConnectionStarting =
LoggerMessage.Define<DateTime>(LogLevel.Debug, 0, "{time}: Starting connection.");
private static readonly Action<ILogger, DateTime, string, Exception> _httpConnectionClosed =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 1, "{time}: Connection Id {connectionId}: Connection was closed from a different thread.");
private static readonly Action<ILogger, DateTime, string, string, Uri, Exception> _startingTransport =
LoggerMessage.Define<DateTime, string, string, Uri>(LogLevel.Debug, 2, "{time}: Connection Id {connectionId}: Starting transport '{transport}' with Url: {url}.");
private static readonly Action<ILogger, DateTime, string, Exception> _raiseConnected =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 3, "{time}: Connection Id {connectionId}: Raising Connected event.");
private static readonly Action<ILogger, DateTime, string, Exception> _processRemainingMessages =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 4, "{time}: Connection Id {connectionId}: Ensuring all outstanding messages are processed.");
private static readonly Action<ILogger, DateTime, string, Exception> _drainEvents =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 5, "{time}: Connection Id {connectionId}: Draining event queue.");
private static readonly Action<ILogger, DateTime, string, Exception> _raiseClosed =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 6, "{time}: Connection Id {connectionId}: Raising Closed event.");
private static readonly Action<ILogger, DateTime, Uri, Exception> _establishingConnection =
LoggerMessage.Define<DateTime, Uri>(LogLevel.Debug, 7, "{time}: Establishing Connection at: {url}.");
private static readonly Action<ILogger, DateTime, Uri, Exception> _errorWithNegotiation =
LoggerMessage.Define<DateTime, Uri>(LogLevel.Error, 8, "{time}: Failed to start connection. Error getting negotiation response from '{url}'.");
private static readonly Action<ILogger, DateTime, string, string, Exception> _errorStartingTransport =
LoggerMessage.Define<DateTime, string, string>(LogLevel.Error, 9, "{time}: Connection Id {connectionId}: Failed to start connection. Error starting transport '{transport}'.");
private static readonly Action<ILogger, DateTime, string, Exception> _httpReceiveStarted =
LoggerMessage.Define<DateTime, string>(LogLevel.Trace, 10, "{time}: Connection Id {connectionId}: Beginning receive loop.");
private static readonly Action<ILogger, DateTime, string, Exception> _skipRaisingReceiveEvent =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 11, "{time}: Connection Id {connectionId}: Message received but connection is not connected. Skipping raising Received event.");
private static readonly Action<ILogger, DateTime, string, Exception> _scheduleReceiveEvent =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 12, "{time}: Connection Id {connectionId}: Scheduling raising Received event.");
private static readonly Action<ILogger, DateTime, string, Exception> _raiseReceiveEvent =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 13, "{time}: Connection Id {connectionId}: Raising Received event.");
private static readonly Action<ILogger, DateTime, string, Exception> _failedReadingMessage =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 14, "{time}: Connection Id {connectionId}: Could not read message.");
private static readonly Action<ILogger, DateTime, string, Exception> _errorReceiving =
LoggerMessage.Define<DateTime, string>(LogLevel.Error, 15, "{time}: Connection Id {connectionId}: Error receiving message.");
private static readonly Action<ILogger, DateTime, string, Exception> _endReceive =
LoggerMessage.Define<DateTime, string>(LogLevel.Trace, 16, "{time}: Connection Id {connectionId}: Ending receive loop.");
private static readonly Action<ILogger, DateTime, string, Exception> _sendingMessage =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 17, "{time}: Connection Id {connectionId}: Sending message.");
private static readonly Action<ILogger, DateTime, string, Exception> _stoppingClient =
LoggerMessage.Define<DateTime, string>(LogLevel.Information, 18, "{time}: Connection Id {connectionId}: Stopping client.");
public static void StartTransport(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_startTransport(logger, DateTime.Now, connectionId, null);
}
}
public static void TransportStopped(this ILogger logger, string connectionId, Exception exception)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_transportStopped(logger, DateTime.Now, connectionId, exception);
}
}
public static void StartReceive(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_startReceive(logger, DateTime.Now, connectionId, null);
}
}
public static void TransportStopping(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_transportStopping(logger, DateTime.Now, connectionId, null);
}
}
public static void WebSocketClosed(this ILogger logger, string connectionId, WebSocketCloseStatus? closeStatus)
{
if (logger.IsEnabled(LogLevel.Information))
{
_webSocketClosed(logger, DateTime.Now, connectionId, closeStatus, null);
}
}
public static void MessageReceived(this ILogger logger, string connectionId, WebSocketMessageType messageType, int count, bool endOfMessage)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_messageReceived(logger, DateTime.Now, connectionId, messageType, count, endOfMessage, null);
}
}
public static void MessageToApp(this ILogger logger, string connectionId, int count)
{
if (logger.IsEnabled(LogLevel.Information))
{
_messageToApp(logger, DateTime.Now, connectionId, count, null);
}
}
public static void ReceiveCanceled(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_receiveCanceled(logger, DateTime.Now, connectionId, null);
}
}
public static void ReceiveStopped(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_receiveStopped(logger, DateTime.Now, connectionId, null);
}
}
public static void SendStarted(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_sendStarted(logger, DateTime.Now, connectionId, null);
}
}
public static void ReceivedFromApp(this ILogger logger, string connectionId, int count)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_receivedFromApp(logger, DateTime.Now, connectionId, count, null);
}
}
public static void SendMessageCanceled(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_sendMessageCanceled(logger, DateTime.Now, connectionId, null);
}
}
public static void ErrorSendingMessage(this ILogger logger, string connectionId, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorSendingMessage(logger, DateTime.Now, connectionId, exception);
}
}
public static void SendCanceled(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_sendCanceled(logger, DateTime.Now, connectionId, null);
}
}
public static void SendStopped(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_sendStopped(logger, DateTime.Now, connectionId, null);
}
}
public static void ClosingWebSocket(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_closingWebSocket(logger, DateTime.Now, connectionId, null);
}
}
public static void ClosingWebSocketFailed(this ILogger logger, string connectionId, Exception exception)
{
if (logger.IsEnabled(LogLevel.Information))
{
_closingWebSocketFailed(logger, DateTime.Now, connectionId, exception);
}
}
public static void SendingMessages(this ILogger logger, string connectionId, int count, Uri url)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_sendingMessages(logger, DateTime.Now, connectionId, count, url, null);
}
}
public static void SentSuccessfully(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_sentSuccessfully(logger, DateTime.Now, connectionId, null);
}
}
public static void NoMessages(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_noMessages(logger, DateTime.Now, connectionId, null);
}
}
public static void ErrorSending(this ILogger logger, string connectionId, Uri url, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorSending(logger, DateTime.Now, connectionId, url, exception);
}
}
public static void EventStreamEnded(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_eventStreamEnded(logger, DateTime.Now, connectionId, null);
}
}
public static void ClosingConnection(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_closingConnection(logger, DateTime.Now, connectionId, null);
}
}
public static void ReceivedMessages(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_receivedMessages(logger, DateTime.Now, connectionId, null);
}
}
public static void ErrorPolling(this ILogger logger, string connectionId, Uri pollUrl, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorPolling(logger, DateTime.Now, connectionId, pollUrl, exception);
}
}
public static void HttpConnectionStarting(this ILogger logger)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_httpConnectionStarting(logger, DateTime.Now, null);
}
}
public static void HttpConnectionClosed(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_httpConnectionClosed(logger, DateTime.Now, connectionId, null);
}
}
public static void StartingTransport(this ILogger logger, string connectionId, string transport, Uri url)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_startingTransport(logger, DateTime.Now, connectionId, transport, url, null);
}
}
public static void RaiseConnected(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_raiseConnected(logger, DateTime.Now, connectionId, null);
}
}
public static void ProcessRemainingMessages(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_processRemainingMessages(logger, DateTime.Now, connectionId, null);
}
}
public static void DrainEvents(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_drainEvents(logger, DateTime.Now, connectionId, null);
}
}
public static void RaiseClosed(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_raiseClosed(logger, DateTime.Now, connectionId, null);
}
}
public static void EstablishingConnection(this ILogger logger, Uri url)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_establishingConnection(logger, DateTime.Now, url, null);
}
}
public static void ErrorWithNegotiation(this ILogger logger, Uri url, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorWithNegotiation(logger, DateTime.Now, url, exception);
}
}
public static void ErrorStartingTransport(this ILogger logger, string connectionId, string transport, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorStartingTransport(logger, DateTime.Now, connectionId, transport, exception);
}
}
public static void HttpReceiveStarted(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Trace))
{
_httpReceiveStarted(logger, DateTime.Now, connectionId, null);
}
}
public static void SkipRaisingReceiveEvent(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_skipRaisingReceiveEvent(logger, DateTime.Now, connectionId, null);
}
}
public static void ScheduleReceiveEvent(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_scheduleReceiveEvent(logger, DateTime.Now, connectionId, null);
}
}
public static void RaiseReceiveEvent(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_raiseReceiveEvent(logger, DateTime.Now, connectionId, null);
}
}
public static void FailedReadingMessage(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_failedReadingMessage(logger, DateTime.Now, connectionId, null);
}
}
public static void ErrorReceiving(this ILogger logger, string connectionId, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_errorReceiving(logger, DateTime.Now, connectionId, exception);
}
}
public static void EndReceive(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Trace))
{
_endReceive(logger, DateTime.Now, connectionId, null);
}
}
public static void SendingMessage(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_sendingMessage(logger, DateTime.Now, connectionId, null);
}
}
public static void StoppingClient(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Information))
{
_stoppingClient(logger, DateTime.Now, connectionId, null);
}
}
}
}

27
src/Microsoft.AspNetCore.Sockets.Client.Http/LongPollingTransport.cs
View File

@ -7,6 +7,7 @@ using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Channels;
using Microsoft.AspNetCore.Sockets.Client.Internal;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
@ -19,6 +20,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
private Channel<byte[], SendMessage> _application;
private Task _sender;
private Task _poller;
private string _connectionId;
private readonly CancellationTokenSource _transportCts = new CancellationTokenSource();
@ -34,20 +36,20 @@ namespace Microsoft.AspNetCore.Sockets.Client
_logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<LongPollingTransport>();
}
public Task StartAsync(Uri url, Channel<byte[], SendMessage> application)
public Task StartAsync(Uri url, Channel<byte[], SendMessage> application, string connectionId)
{
_logger.LogInformation("Starting {0}", nameof(LongPollingTransport));
_connectionId = connectionId;
_logger.StartTransport(_connectionId);
_application = application;
// Start sending and polling (ask for binary if the server supports it)
_poller = Poll(url, _transportCts.Token);
_sender = SendUtils.SendMessages(url, _application, _httpClient, _transportCts, _logger);
_sender = SendUtils.SendMessages(url, _application, _httpClient, _transportCts, _logger, _connectionId);
Running = Task.WhenAll(_sender, _poller).ContinueWith(t =>
{
_logger.LogDebug("Transport stopped. Exception: '{0}'", t.Exception?.InnerException);
_logger.TransportStopped(_connectionId, t.Exception?.InnerException);
_application.Out.TryComplete(t.IsFaulted ? t.Exception.InnerException : null);
return t;
}).Unwrap();
@ -57,7 +59,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
public async Task StopAsync()
{
_logger.LogInformation("Transport {0} is stopping", nameof(LongPollingTransport));
_logger.TransportStopping(_connectionId);
_transportCts.Cancel();
@ -69,13 +71,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
// exceptions have been handled in the Running task continuation by closing the channel with the exception
}
_logger.LogInformation("Transport {0} stopped", nameof(LongPollingTransport));
}
private async Task Poll(Uri pollUrl, CancellationToken cancellationToken)
{
_logger.LogInformation("Starting the receive loop");
_logger.StartReceive(_connectionId);
try
{
while (!cancellationToken.IsCancellationRequested)
@ -88,14 +88,14 @@ namespace Microsoft.AspNetCore.Sockets.Client
if (response.StatusCode == HttpStatusCode.NoContent || cancellationToken.IsCancellationRequested)
{
_logger.LogDebug("The server is closing the connection");
_logger.ClosingConnection(_connectionId);
// Transport closed or polling stopped, we're done
break;
}
else
{
_logger.LogDebug("Received messages from the server");
_logger.ReceivedMessages(_connectionId);
// Until Pipeline starts natively supporting BytesReader, this is the easiest way to do this.
var payload = await response.Content.ReadAsByteArrayAsync();
@ -115,17 +115,18 @@ namespace Microsoft.AspNetCore.Sockets.Client
catch (OperationCanceledException)
{
// transport is being closed
_logger.ReceiveCanceled(_connectionId);
}
catch (Exception ex)
{
_logger.LogError("Error while polling '{0}': {1}", pollUrl, ex);
_logger.ErrorPolling(_connectionId, pollUrl, ex);
throw;
}
finally
{
// Make sure the send loop is terminated
_transportCts.Cancel();
_logger.LogInformation("Receive loop stopped");
_logger.ReceiveStopped(_connectionId);
}
}
}

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

@ -9,6 +9,7 @@ using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Channels;
using Microsoft.AspNetCore.Sockets.Client.Internal;
using Microsoft.Extensions.Logging;
namespace Microsoft.AspNetCore.Sockets.Client
@ -18,9 +19,10 @@ namespace Microsoft.AspNetCore.Sockets.Client
private static readonly string DefaultUserAgent = "Microsoft.AspNetCore.SignalR.Client/0.0.0";
public static readonly ProductInfoHeaderValue DefaultUserAgentHeader = ProductInfoHeaderValue.Parse(DefaultUserAgent);
public static async Task SendMessages(Uri sendUrl, Channel<byte[], SendMessage> application, HttpClient httpClient, CancellationTokenSource transportCts, ILogger logger)
public static async Task SendMessages(Uri sendUrl, Channel<byte[], SendMessage> application, HttpClient httpClient,
CancellationTokenSource transportCts, ILogger logger, string connectionId)
{
logger.LogInformation("Starting the send loop");
logger.SendStarted(connectionId);
IList<SendMessage> messages = null;
try
{
@ -35,7 +37,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
if (messages.Count > 0)
{
logger.LogDebug("Sending {0} message(s) to the server using url: {1}", messages.Count, sendUrl);
logger.SendingMessages(connectionId, messages.Count, sendUrl);
// Send them in a single post
var request = new HttpRequestMessage(HttpMethod.Post, sendUrl);
@ -61,7 +63,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
var response = await httpClient.SendAsync(request);
response.EnsureSuccessStatusCode();
logger.LogDebug("Message(s) sent successfully");
logger.SentSuccessfully(connectionId);
foreach (var message in messages)
{
message.SendResult?.TrySetResult(null);
@ -69,7 +71,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
else
{
logger.LogDebug("No messages in batch to send");
logger.NoMessages(connectionId);
}
}
}
@ -84,10 +86,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
message.SendResult?.TrySetCanceled();
}
}
logger.SendCanceled(connectionId);
}
catch (Exception ex)
{
logger.LogError("Error while sending to '{0}': {1}", sendUrl, ex);
logger.ErrorSending(connectionId, sendUrl, ex);
if (messages != null)
{
foreach (var message in messages)
@ -104,7 +107,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
transportCts.Cancel();
}
logger.LogInformation("Send loop stopped");
logger.SendStopped(connectionId);
}
}
}

21
src/Microsoft.AspNetCore.Sockets.Client.Http/ServerSentEventsTransport.cs
View File

@ -8,6 +8,7 @@ using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Channels;
using Microsoft.AspNetCore.Sockets.Client.Internal;
using Microsoft.AspNetCore.Sockets.Internal.Formatters;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
@ -20,6 +21,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
private readonly ILogger _logger;
private readonly CancellationTokenSource _transportCts = new CancellationTokenSource();
private readonly ServerSentEventsMessageParser _parser = new ServerSentEventsMessageParser();
private string _connectionId;
private Channel<byte[], SendMessage> _application;
@ -40,20 +42,18 @@ namespace Microsoft.AspNetCore.Sockets.Client
_logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<ServerSentEventsTransport>();
}
public Task StartAsync(Uri url, Channel<byte[], SendMessage> application)
public Task StartAsync(Uri url, Channel<byte[], SendMessage> application, string connectionId)
{
_logger.LogInformation("Starting {transportName}", nameof(ServerSentEventsTransport));
_connectionId = connectionId;
_logger.StartTransport(_connectionId);
_application = application;
var sendTask = SendUtils.SendMessages(url, _application, _httpClient, _transportCts, _logger);
var sendTask = SendUtils.SendMessages(url, _application, _httpClient, _transportCts, _logger, _connectionId);
var receiveTask = OpenConnection(_application, url, _transportCts.Token);
Running = Task.WhenAll(sendTask, receiveTask).ContinueWith(t =>
{
if (t.Exception != null)
{
_logger.LogError(0, t.Exception.InnerException, "Transport stopped");
}
_logger.TransportStopped(_connectionId, t.Exception?.InnerException);
_application.Out.TryComplete(t.IsFaulted ? t.Exception.InnerException : null);
return t;
@ -64,7 +64,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
private async Task OpenConnection(Channel<byte[], SendMessage> application, Uri url, CancellationToken cancellationToken)
{
_logger.LogInformation("Starting receive loop");
_logger.StartReceive(_connectionId);
var request = new HttpRequestMessage(HttpMethod.Get, url);
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
@ -82,7 +82,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
var input = result.Buffer;
if (result.IsCancelled || (input.IsEmpty && result.IsCompleted))
{
_logger.LogDebug("Server-Sent Event Stream ended");
_logger.EventStreamEnded(_connectionId);
break;
}
@ -118,12 +118,13 @@ namespace Microsoft.AspNetCore.Sockets.Client
readCancellationRegistration.Dispose();
_transportCts.Cancel();
stream.Dispose();
_logger.ReceiveStopped(_connectionId);
}
}
public async Task StopAsync()
{
_logger.LogInformation("Transport {transportName} is stopping", nameof(ServerSentEventsTransport));
_logger.TransportStopping(_connectionId);
_transportCts.Cancel();
_application.Out.TryComplete();
await Running;

50
src/Microsoft.AspNetCore.Sockets.Client.Http/WebSocketsTransport.cs
View File

@ -8,6 +8,7 @@ using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Channels;
using Microsoft.AspNetCore.Sockets.Client.Internal;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
@ -19,6 +20,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
private Channel<byte[], SendMessage> _application;
private readonly CancellationTokenSource _transportCts = new CancellationTokenSource();
private readonly ILogger _logger;
private string _connectionId;
public WebSocketsTransport()
: this(null)
@ -27,15 +29,13 @@ namespace Microsoft.AspNetCore.Sockets.Client
public WebSocketsTransport(ILoggerFactory loggerFactory)
{
_logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger(nameof(WebSocketsTransport));
_logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<WebSocketsTransport>();
}
public Task Running { get; private set; } = Task.CompletedTask;
public async Task StartAsync(Uri url, Channel<byte[], SendMessage> application)
public async Task StartAsync(Uri url, Channel<byte[], SendMessage> application, string connectionId)
{
_logger.LogInformation("Starting {0}", nameof(WebSocketsTransport));
if (url == null)
{
throw new ArgumentNullException(nameof(url));
@ -47,7 +47,9 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
_application = application;
_connectionId = connectionId;
_logger.StartTransport(_connectionId);
await Connect(url);
var sendTask = SendMessages(url);
var receiveTask = ReceiveMessages(url);
@ -56,16 +58,15 @@ namespace Microsoft.AspNetCore.Sockets.Client
// https://github.com/SignalR/SignalR/blob/1fba14fa3437e24c204dfaf8a18db3fce8acad3c/src/Microsoft.AspNet.SignalR.Core/Owin/WebSockets/WebSocketHandler.cs#L248-L251
Running = Task.WhenAll(sendTask, receiveTask).ContinueWith(t =>
{
_logger.LogDebug("Transport stopped. Exception: '{0}'", t.Exception?.InnerException);
_application.Out.TryComplete(t.IsFaulted ? t.Exception.InnerException : null);
return t;
_logger.TransportStopped(_connectionId, t.Exception?.InnerException);
_application.Out.TryComplete(t.IsFaulted ? t.Exception.InnerException : null);
return t;
}).Unwrap();
}
private async Task ReceiveMessages(Uri pollUrl)
{
_logger.LogInformation("Starting receive loop");
_logger.StartReceive(_connectionId);
try
{
@ -83,7 +84,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
receiveResult = await _webSocket.ReceiveAsync(buffer, _transportCts.Token);
if (receiveResult.MessageType == WebSocketMessageType.Close)
{
_logger.LogInformation("Websocket closed by the server. Close status {0}", receiveResult.CloseStatus);
_logger.WebSocketClosed(_connectionId, receiveResult.CloseStatus);
_application.Out.Complete(
receiveResult.CloseStatus == WebSocketCloseStatus.NormalClosure
@ -93,8 +94,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
return;
}
_logger.LogDebug("Message received. Type: {0}, size: {1}, EndOfMessage: {2}",
receiveResult.MessageType.ToString(), receiveResult.Count, receiveResult.EndOfMessage);
_logger.MessageReceived(_connectionId, receiveResult.MessageType, receiveResult.Count, receiveResult.EndOfMessage);
var truncBuffer = new ArraySegment<byte>(buffer.Array, 0, receiveResult.Count);
incomingMessage.Add(truncBuffer);
@ -119,7 +119,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
Buffer.BlockCopy(incomingMessage[0].Array, incomingMessage[0].Offset, messageBuffer, 0, incomingMessage[0].Count);
}
_logger.LogInformation("Passing message to application. Payload size: {0}", messageBuffer.Length);
_logger.MessageToApp(_connectionId, messageBuffer.Length);
while (await _application.Out.WaitToWriteAsync(_transportCts.Token))
{
if (_application.Out.TryWrite(messageBuffer))
@ -132,18 +132,18 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
catch (OperationCanceledException)
{
_logger.LogDebug("Receive loop canceled");
_logger.ReceiveCanceled(_connectionId);
}
finally
{
_logger.LogInformation("Receive loop stopped");
_logger.ReceiveStopped(_connectionId);
_transportCts.Cancel();
}
}
private async Task SendMessages(Uri sendUrl)
{
_logger.LogInformation("Starting the send loop");
_logger.SendStarted(_connectionId);
try
{
@ -153,7 +153,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
try
{
_logger.LogDebug("Received message from application. Payload size: {1}", message.Payload.Length);
_logger.ReceivedFromApp(_connectionId, message.Payload.Length);
await _webSocket.SendAsync(new ArraySegment<byte>(message.Payload), WebSocketMessageType.Text, true, _transportCts.Token);
@ -161,14 +161,14 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
catch (OperationCanceledException)
{
_logger.LogInformation("Sending a message canceled.");
_logger.SendMessageCanceled(_connectionId);
message.SendResult.SetCanceled();
await CloseWebSocket();
break;
}
catch (Exception ex)
{
_logger.LogError("Error while sending a message {0}", ex.Message);
_logger.ErrorSendingMessage(_connectionId, ex);
message.SendResult.SetException(ex);
await CloseWebSocket();
throw;
@ -178,11 +178,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
}
catch (OperationCanceledException)
{
_logger.LogDebug("Send loop canceled");
_logger.SendCanceled(_connectionId);
}
finally
{
_logger.LogInformation("Send loop stopped");
_logger.SendStopped(_connectionId);
_transportCts.Cancel();
}
}
@ -204,7 +204,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
public async Task StopAsync()
{
_logger.LogInformation("Transport {0} is stopping", nameof(WebSocketsTransport));
_logger.TransportStopping(_connectionId);
await CloseWebSocket();
_webSocket.Dispose();
@ -217,8 +217,6 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
// exceptions have been handled in the Running task continuation by closing the channel with the exception
}
_logger.LogInformation("Transport {0} stopped", nameof(WebSocketsTransport));
}
private async Task CloseWebSocket()
@ -229,7 +227,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
// while the webSocket is being closed due to an error.
if (_webSocket.State != WebSocketState.Closed)
{
_logger.LogInformation("Closing webSocket");
_logger.ClosingWebSocket(_connectionId);
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
}
}
@ -237,7 +235,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
// This is benign - the exception can happen due to the race described above because we would
// try closing the webSocket twice.
_logger.LogInformation("Closing webSocket failed with {0}", ex);
_logger.ClosingWebSocketFailed(_connectionId, ex);
}
}
}

2
src/Microsoft.AspNetCore.Sockets.Http/Internal/SocketLoggerExtensions.cs → src/Microsoft.AspNetCore.Sockets.Http/Internal/SocketHttpLoggerExtensions.cs
View File

@ -7,7 +7,7 @@ using Microsoft.Extensions.Logging;
namespace Microsoft.AspNetCore.Sockets.Internal
{
public static class SocketLoggerExtensions
internal static class SocketHttpLoggerExtensions
{
// Category: LongPollingTransport
private static readonly Action<ILogger, DateTime, string, string, Exception> _longPolling204 =

8
src/Microsoft.AspNetCore.Sockets/ConnectionManager.cs
View File

@ -51,6 +51,8 @@ namespace Microsoft.AspNetCore.Sockets
{
var id = MakeNewConnectionId();
_logger.CreatedNewConnection(id);
var transportToApplication = Channel.CreateUnbounded<byte[]>();
var applicationToTransport = Channel.CreateUnbounded<byte[]>();
@ -68,7 +70,7 @@ namespace Microsoft.AspNetCore.Sockets
if (_connections.TryRemove(id, out _))
{
// Remove the connection completely
_logger.LogDebug("Removing {connectionId} from the list of connections", id);
_logger.RemovedConnection(id);
}
}
@ -124,7 +126,7 @@ namespace Microsoft.AspNetCore.Sockets
c.Value.Lock.Release();
}
// Once the decision has been made to to dispose we don't check the status again
// Once the decision has been made to dispose we don't check the status again
if (status == DefaultConnectionContext.ConnectionStatus.Inactive && (DateTimeOffset.UtcNow - lastSeenUtc).TotalSeconds > 5)
{
var ignore = DisposeAndRemoveAsync(c.Value);
@ -174,7 +176,7 @@ namespace Microsoft.AspNetCore.Sockets
}
catch (Exception ex)
{
_logger.LogError(0, ex, "Failed disposing connection {connectionId}", connection.ConnectionId);
_logger.FailedDispose(connection.ConnectionId, ex);
}
finally
{

45
src/Microsoft.AspNetCore.Sockets/Internal/SocketLoggerExtensions.cs Normal file
View File

@ -0,0 +1,45 @@
// 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 Microsoft.Extensions.Logging;
namespace Microsoft.AspNetCore.Sockets.Internal
{
internal static class SocketLoggerExtensions
{
// Category: ConnectionManager
private static readonly Action<ILogger, DateTime, string, Exception> _createdNewConnection =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 0, "{time}: ConnectionId {connectionId}: New connection created.");
private static readonly Action<ILogger, DateTime, string, Exception> _removedConnection =
LoggerMessage.Define<DateTime, string>(LogLevel.Debug, 1, "{time}: ConnectionId {connectionId}: Removing connection from the list of connections.");
private static readonly Action<ILogger, DateTime, string, Exception> _failedDispose =
LoggerMessage.Define<DateTime, string>(LogLevel.Error, 2, "{time}: ConnectionId {connectionId}: Failed disposing connection.");
public static void CreatedNewConnection(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_createdNewConnection(logger, DateTime.Now, connectionId, null);
}
}
public static void RemovedConnection(this ILogger logger, string connectionId)
{
if (logger.IsEnabled(LogLevel.Debug))
{
_removedConnection(logger, DateTime.Now, connectionId, null);
}
}
public static void FailedDispose(this ILogger logger, string connectionId, Exception exception)
{
if (logger.IsEnabled(LogLevel.Error))
{
_failedDispose(logger, DateTime.Now, connectionId, exception);
}
}
}
}

12
test/Microsoft.AspNetCore.SignalR.Client.Tests/ConnectionTests.cs
View File

@ -154,7 +154,7 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests
releaseDisposeTcs.SetResult(null);
await disposeTask.OrTimeout();
transport.Verify(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>()), Times.Never);
transport.Verify(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>(), It.IsAny<string>()), Times.Never);
}
[Fact]
@ -241,7 +241,7 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests
});
var mockTransport = new Mock<ITransport>();
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>()))
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>(), It.IsAny<string>()))
.Returns(Task.FromException(new InvalidOperationException("Transport failed to start")));
@ -350,8 +350,8 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests
var mockTransport = new Mock<ITransport>();
Channel<byte[], SendMessage> channel = null;
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>()))
.Returns<Uri, Channel<byte[], SendMessage>>((url, c) =>
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>(), It.IsAny<string>()))
.Returns<Uri, Channel<byte[], SendMessage>, string>((url, c, id) =>
{
channel = c;
return Task.CompletedTask;
@ -396,8 +396,8 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests
var mockTransport = new Mock<ITransport>();
Channel<byte[], SendMessage> channel = null;
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>()))
.Returns<Uri, Channel<byte[], SendMessage>>((url, c) =>
mockTransport.Setup(t => t.StartAsync(It.IsAny<Uri>(), It.IsAny<Channel<byte[], SendMessage>>(), It.IsAny<string>()))
.Returns<Uri, Channel<byte[], SendMessage>, string>((url, c, id) =>
{
channel = c;
return Task.CompletedTask;

16
test/Microsoft.AspNetCore.SignalR.Client.Tests/LongPollingTransportTests.cs
View File

@ -44,7 +44,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
transportActiveTask = longPollingTransport.Running;
@ -80,7 +80,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = ChannelConnection.Create(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
await longPollingTransport.Running.OrTimeout();
Assert.True(transportToConnection.In.Completion.IsCompleted);
@ -133,7 +133,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
var data = await transportToConnection.In.ReadAllAsync().OrTimeout();
await longPollingTransport.Running.OrTimeout();
@ -169,7 +169,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
var exception =
await Assert.ThrowsAsync<HttpRequestException>(async () => await transportToConnection.In.Completion.OrTimeout());
@ -205,7 +205,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
await connectionToTransport.Out.WriteAsync(new SendMessage());
@ -246,7 +246,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
connectionToTransport.Out.Complete();
@ -297,7 +297,7 @@ namespace Microsoft.AspNetCore.Client.Tests
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
// Start the transport
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
// Wait for the transport to finish
await longPollingTransport.Running.OrTimeout();
@ -362,7 +362,7 @@ namespace Microsoft.AspNetCore.Client.Tests
await connectionToTransport.Out.WriteAsync(new SendMessage(Encoding.UTF8.GetBytes("World"), tcs2)).OrTimeout();
// Start the transport
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection);
await longPollingTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty);
connectionToTransport.Out.Complete();

2
test/Microsoft.AspNetCore.SignalR.Client.Tests/ServerSentEventsTransportTests.cs
View File

@ -49,7 +49,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
var connectionToTransport = Channel.CreateUnbounded<SendMessage>();
var transportToConnection = Channel.CreateUnbounded<byte[]>();
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
await sseTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection).OrTimeout();
await sseTransport.StartAsync(new Uri("http://fakeuri.org"), channelConnection, connectionId: string.Empty).OrTimeout();
await eventStreamTcs.Task.OrTimeout();
await sseTransport.StopAsync().OrTimeout();

15
test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs
View File

@ -129,6 +129,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
await closeTcs.Task.OrTimeout();
}
catch (Exception ex)
{
logger.LogInformation(ex, "Test threw exception");
throw;
}
finally
{
logger.LogInformation("Disposing Connection");
@ -182,6 +187,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
Assert.Equal(message, Encoding.UTF8.GetString(receivedData));
logger.LogInformation("Completed receive");
}
catch (Exception ex)
{
logger.LogInformation(ex, "Test threw exception");
throw;
}
finally
{
logger.LogInformation("Disposing Connection");
@ -240,6 +250,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
await closeTcs.Task.OrTimeout();
}
catch (Exception ex)
{
logger.LogInformation(ex, "Test threw exception");
throw;
}
finally
{
logger.LogInformation("Disposing Connection");

2
test/Microsoft.AspNetCore.SignalR.Tests/Internal/DefaultHubProtocolResolverTests.cs
View File

@ -37,7 +37,7 @@ namespace Microsoft.AspNetCore.SignalR.Common.Protocol.Tests
Assert.Equal($"The protocol '{protocolName ?? "(null)"}' is not supported.", exception.Message);
}
public static IEnumerable<object[]> HubProtocols() =>
public static IEnumerable<object[]> HubProtocols =>
new[]
{
new object[] { new JsonHubProtocol(new JsonSerializer()) },

6
test/Microsoft.AspNetCore.SignalR.Tests/WebSocketsTransportTests.cs
View File

@ -40,7 +40,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
var webSocketsTransport = new WebSocketsTransport(loggerFactory);
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection).OrTimeout();
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection, connectionId: string.Empty).OrTimeout();
await webSocketsTransport.StopAsync().OrTimeout();
await webSocketsTransport.Running.OrTimeout();
}
@ -57,7 +57,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
var webSocketsTransport = new WebSocketsTransport(loggerFactory);
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection);
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection, connectionId: string.Empty);
connectionToTransport.Out.TryComplete();
await webSocketsTransport.Running.OrTimeout();
}
@ -74,7 +74,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
var channelConnection = new ChannelConnection<SendMessage, byte[]>(connectionToTransport, transportToConnection);
var webSocketsTransport = new WebSocketsTransport(loggerFactory);
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection);
await webSocketsTransport.StartAsync(new Uri(_serverFixture.WebSocketsUrl + "/echo"), channelConnection, connectionId: string.Empty);
var sendTcs = new TaskCompletionSource<object>();
connectionToTransport.Out.TryWrite(new SendMessage(new byte[] { 0x42 }, sendTcs));