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

De-statificying HubConnection 

Wiring and raising events correctly
This commit is contained in:
moozzyk 2017-02-09 17:15:18 -08:00
parent d4fbdd055a
commit a3e847e6cc
8 changed files with 260 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
SignalR.sln
View File

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

4
samples/ClientSample/HubSample.cs
View File

@ -30,9 +30,9 @@ namespace ClientSample
{
logger.LogInformation("Connecting to {0}", baseUrl);
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri(baseUrl),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri(baseUrl), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
logger.LogInformation("Connected to {0}", baseUrl);
var cts = new CancellationTokenSource();

73
src/Microsoft.AspNetCore.SignalR.Client/HubConnection.cs
View File

@ -33,15 +33,52 @@ namespace Microsoft.AspNetCore.SignalR.Client
private int _nextId = 0;
private HubConnection(Connection connection, IInvocationAdapter adapter, ILogger logger)
public event Action Connected
{
_binder = new HubBinder(this);
_connection = connection;
_adapter = adapter;
_logger = logger;
add { _connection.Connected += value; }
remove { _connection.Connected -= value; }
}
public event Action<Exception> Closed
{
add { _connection.Closed += value; }
remove { _connection.Closed -= value; }
}
public HubConnection(Uri url, IInvocationAdapter adapter, ILoggerFactory loggerFactory)
{
// TODO: loggerFactory shouldn't be required
if (loggerFactory == null)
{
throw new ArgumentNullException(nameof(loggerFactory));
}
_binder = new HubBinder(this);
_connection = new Connection(url, loggerFactory);
_adapter = adapter;
_logger = loggerFactory.CreateLogger<HubConnection>();
// TODO HIGH: Need to subscribe to events before starting connection. Requires converting HubConnection
_connection.Received += OnDataReceived;
_connection.Closed += Shutdown;
}
public Task StartAsync() => StartAsync(null, null);
public Task StartAsync(HttpClient httpClient) => StartAsync(null, httpClient);
public Task StartAsync(ITransport transport) => StartAsync(transport, null);
public async Task StartAsync(ITransport transport, HttpClient httpClient)
{
await _connection.StartAsync(transport, httpClient);
}
public async Task StopAsync()
{
await _connection.StopAsync();
}
public void Dispose()
{
_connection.Dispose();
}
// TODO: Client return values/tasks?
@ -59,7 +96,7 @@ namespace Microsoft.AspNetCore.SignalR.Client
public Task<object> Invoke(string methodName, Type returnType, params object[] args) => Invoke(methodName, returnType, CancellationToken.None, args);
public async Task<object> Invoke(string methodName, Type returnType, CancellationToken cancellationToken, params object[] args)
{
// TODO: we should reject calls to here after the connection is "done" (e.g. sending an invocation failed)
// TODO: we should reject calls to here after the connection is "done" or has not been started (e.g. sending an invocation failed)
_logger.LogTrace("Preparing invocation of '{0}', with return type '{1}' and {2} args", methodName, returnType.AssemblyQualifiedName, args.Length);
@ -97,33 +134,14 @@ namespace Microsoft.AspNetCore.SignalR.Client
_logger.LogInformation("Sending Invocation #{0}", descriptor.Id);
// TODO: Format.Text - who, where and when decides about the format of outgoing messages
// TODO HIGH: Handle return value/Exception from SendAsync
await _connection.SendAsync(ms.ToArray(), MessageType.Text, cancellationToken);
_logger.LogInformation("Sending Invocation #{0} complete", descriptor.Id);
// Return the completion task. It will be completed by ReceiveMessages when the response is received.
return await irq.Completion.Task;
}
// TODO HIGH - need StopAsync
public void Dispose()
{
_connection.Dispose();
}
// TODO: Clean up the API here. Negotiation of format would be better than providing an adapter instance. Similarly, we should not require a logger factory
public static Task<HubConnection> ConnectAsync(Uri url, IInvocationAdapter adapter, ITransport transport, ILoggerFactory loggerFactory) => ConnectAsync(url, adapter, transport, new HttpClient(), loggerFactory);
public static async Task<HubConnection> ConnectAsync(Uri url, IInvocationAdapter adapter, ITransport transport, HttpClient httpClient, ILoggerFactory loggerFactory)
{
// Connect the underlying connection
var connection = new Connection(url, loggerFactory);
await connection.StartAsync(transport, httpClient);
return new HubConnection(connection, adapter, loggerFactory.CreateLogger<HubConnection>());
}
private async void OnDataReceived(byte[] data, MessageType messageType)
{
var message
@ -147,7 +165,6 @@ namespace Microsoft.AspNetCore.SignalR.Client
}
}
// TODO HIGH!
private void Shutdown(Exception ex = null)
{
_logger.LogTrace("Shutting down connection");

11
src/Microsoft.AspNetCore.Sockets.Client/Connection.cs
View File

@ -42,10 +42,11 @@ namespace Microsoft.AspNetCore.Sockets.Client
_logger = _loggerFactory.CreateLogger<Connection>();
}
public Task StartAsync() => StartAsync(null, null);
public Task StartAsync(HttpClient httpClient) => StartAsync(null, httpClient);
public Task StartAsync() => StartAsync(transport: null, httpClient: null);
public Task StartAsync(HttpClient httpClient) => StartAsync(transport: null, httpClient: httpClient);
public Task StartAsync(ITransport transport) => StartAsync(transport, null);
// TODO HIGH: Fix a race when the connection is being stopped/disposed when start has not finished running
public async Task StartAsync(ITransport transport, HttpClient httpClient)
{
// TODO: make transport optional
@ -124,8 +125,8 @@ namespace Microsoft.AspNetCore.Sockets.Client
var applicationSide = new ChannelConnection<Message>(transportToApplication, applicationToTransport);
_transportChannel = new ChannelConnection<Message>(applicationToTransport, transportToApplication);
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Input.Completion.ContinueWith(t =>
var ignore = Input.Completion.ContinueWith(t =>
{
Interlocked.Exchange(ref _connectionState, ConnectionState.Disconnected);
@ -136,7 +137,6 @@ namespace Microsoft.AspNetCore.Sockets.Client
closedEventHandler(t.IsFaulted ? t.Exception.InnerException : null);
}
});
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
// Start the transport, giving it one end of the pipeline
try
@ -162,6 +162,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
{
using (message)
{
// Do not "simplify" - events can be removed from a different thread
var receivedEventHandler = Received;
if (receivedEventHandler != null)
{

1
src/Microsoft.AspNetCore.Sockets.Client/NullLoggerFactory.cs
View File

@ -1,7 +1,6 @@
// 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;
using Microsoft.Extensions.Logging.Abstractions;

29
test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs
View File

@ -1,8 +1,6 @@
// 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.Threading.Tasks;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.SignalR.Tests.Common;
@ -10,6 +8,8 @@ using Microsoft.AspNetCore.Sockets.Client;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System;
using System.Threading.Tasks;
using Xunit;
namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
@ -51,9 +51,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
using (var httpClient = _testServer.CreateClient())
{
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri("http://test/hubs"),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri("http://test/hubs"), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
var result = await connection.Invoke<string>("HelloWorld");
Assert.Equal("Hello World!", result);
@ -70,9 +71,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
using (var httpClient = _testServer.CreateClient())
{
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri("http://test/hubs"),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri("http://test/hubs"), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
var result = await connection.Invoke<string>("Echo", originalMessage);
Assert.Equal(originalMessage, result);
@ -89,9 +91,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
using (var httpClient = _testServer.CreateClient())
{
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri("http://test/hubs"),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri("http://test/hubs"), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
var result = await connection.Invoke<string>("echo", originalMessage);
Assert.Equal(originalMessage, result);
@ -108,9 +111,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
using (var httpClient = _testServer.CreateClient())
{
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri("http://test/hubs"),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri("http://test/hubs"), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
var tcs = new TaskCompletionSource<string>();
connection.On("Echo", new[] { typeof(string) }, a =>
{
@ -132,9 +136,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
using (var httpClient = _testServer.CreateClient())
{
var transport = new LongPollingTransport(httpClient, loggerFactory);
using (var connection = await HubConnection.ConnectAsync(new Uri("http://test/hubs"),
new JsonNetInvocationAdapter(), transport, httpClient, loggerFactory))
using (var connection = new HubConnection(new Uri("http://test/hubs"), new JsonNetInvocationAdapter(), loggerFactory))
{
await connection.StartAsync(transport, httpClient);
var ex = await Assert.ThrowsAnyAsync<Exception>(
async () => await connection.Invoke<object>("!@#$%"));

188
test/Microsoft.AspNetCore.Sockets.Client.Tests/HubConnectionTests.cs Normal file
View File

@ -0,0 +1,188 @@
using Microsoft.AspNetCore.Sockets;
using Microsoft.AspNetCore.Sockets.Client;
using Microsoft.Extensions.Logging;
using Moq;
using Moq.Protected;
using System;
using System.Net;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
namespace Microsoft.AspNetCore.SignalR.Client.Tests
{
public class HubConnectionTests
{
[Fact]
public void CannotCreateHubConnectionWithNullUrl()
{
var exception = Assert.Throws<ArgumentNullException>(
() => new HubConnection(null, Mock.Of<IInvocationAdapter>(), Mock.Of<ILoggerFactory>()));
Assert.Equal("url", exception.ParamName);
}
[Fact]
public void CanDisposeNotStartedHubConnection()
{
using (new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), Mock.Of<ILoggerFactory>()))
{ }
}
[Fact]
public async Task CanStopNotStartedHubConnection()
{
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), Mock.Of<ILoggerFactory>()))
{
await hubConnection.StopAsync();
}
}
[Fact]
public async Task CannotStartRunningHubConnection()
{
var mockHttpHandler = new Mock<HttpMessageHandler>();
mockHttpHandler.Protected()
.Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
.Returns<HttpRequestMessage, CancellationToken>(async (request, cancellationToken) =>
{
await Task.Yield();
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(string.Empty) };
});
using (var httpClient = new HttpClient(mockHttpHandler.Object))
using (var longPollingTransport = new LongPollingTransport(httpClient, new LoggerFactory()))
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org/"), Mock.Of<IInvocationAdapter>(), new LoggerFactory()))
{
await hubConnection.StartAsync(longPollingTransport, httpClient);
var exception =
await Assert.ThrowsAsync<InvalidOperationException>(
async () => await hubConnection.StartAsync(longPollingTransport));
Assert.Equal("Cannot start a connection that is not in the Initial state.", exception.Message);
await hubConnection.StopAsync();
}
}
[Fact]
public async Task CannotStartStoppedHubConnection()
{
var mockHttpHandler = new Mock<HttpMessageHandler>();
mockHttpHandler.Protected()
.Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
.Returns<HttpRequestMessage, CancellationToken>(async (request, cancellationToken) =>
{
await Task.Yield();
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(string.Empty) };
});
using (var httpClient = new HttpClient(mockHttpHandler.Object))
using (var longPollingTransport = new LongPollingTransport(httpClient, new LoggerFactory()))
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org/"), Mock.Of<IInvocationAdapter>(), new LoggerFactory()))
{
await hubConnection.StartAsync(longPollingTransport, httpClient);
await hubConnection.StopAsync();
var exception =
await Assert.ThrowsAsync<InvalidOperationException>(
async () => await hubConnection.StartAsync(longPollingTransport));
Assert.Equal("Cannot start a connection that is not in the Initial state.", exception.Message);
}
}
[Fact]
public async Task CannotStartDisposedHubConnection()
{
using (var httpClient = new HttpClient())
using (var longPollingTransport = new LongPollingTransport(httpClient, new LoggerFactory()))
{
var hubConnection = new HubConnection(new Uri("http://fakeuri.org/"), Mock.Of<IInvocationAdapter>(), new LoggerFactory());
hubConnection.Dispose();
var exception =
await Assert.ThrowsAsync<InvalidOperationException>(
async () => await hubConnection.StartAsync(longPollingTransport));
Assert.Equal("Cannot start a connection that is not in the Initial state.", exception.Message);
}
}
[Fact(Skip = "Not implemented")]
public async Task InvokeThrowsIfHubConnectionNotStarted()
{
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), Mock.Of<ILoggerFactory>()))
{
var exception =
await Assert.ThrowsAsync<InvalidOperationException>(async () => await hubConnection.Invoke<int>("test"));
Assert.Equal("Cannot invoke methods on non-started connections.", exception.Message);
}
}
[Fact(Skip = "Not implemented")]
public async Task InvokeThrowsIfHubConnectionDisposed()
{
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), Mock.Of<ILoggerFactory>()))
{
hubConnection.Dispose();
var exception =
await Assert.ThrowsAsync<InvalidOperationException>(async () => await hubConnection.Invoke<int>("test"));
Assert.Equal("Cannot invoke methods on disposed connections.", exception.Message);
}
}
// TODO: If HubConnection takes (I)Connection we could just tests if events are wired up
[Fact]
public async Task HubConnectionConnectedEventRaisedWhenTheClientIsConnected()
{
var mockHttpHandler = new Mock<HttpMessageHandler>();
mockHttpHandler.Protected()
.Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
.Returns<HttpRequestMessage, CancellationToken>(async (request, cancellationToken) =>
{
await Task.Yield();
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(string.Empty) };
});
using (var httpClient = new HttpClient(mockHttpHandler.Object))
using (var longPollingTransport = new LongPollingTransport(httpClient, new LoggerFactory()))
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), new LoggerFactory()))
{
var connectedEventRaised = false;
hubConnection.Connected += () => connectedEventRaised = true;
await hubConnection.StartAsync(longPollingTransport, httpClient);
Assert.True(connectedEventRaised);
}
}
[Fact]
public async Task ClosedEventRaisedWhenTheClientIsStopped()
{
var mockHttpHandler = new Mock<HttpMessageHandler>();
mockHttpHandler.Protected()
.Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
.Returns<HttpRequestMessage, CancellationToken>(async (request, cancellationToken) =>
{
await Task.Yield();
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(string.Empty) };
});
using (var httpClient = new HttpClient(mockHttpHandler.Object))
using (var longPollingTransport = new LongPollingTransport(httpClient, new LoggerFactory()))
using (var hubConnection = new HubConnection(new Uri("http://fakeuri.org"), Mock.Of<IInvocationAdapter>(), new LoggerFactory()))
{
var closedEventTcs = new TaskCompletionSource<Exception>();
hubConnection.Closed += e => closedEventTcs.SetResult(e);
await hubConnection.StartAsync(longPollingTransport, httpClient);
await hubConnection.StopAsync();
Assert.Equal(closedEventTcs.Task, await Task.WhenAny(Task.Delay(1000), closedEventTcs.Task));
// in case of clean disconnect error should be null
Assert.Null(await closedEventTcs.Task);
}
}
}
}

2
test/Microsoft.AspNetCore.Sockets.Client.Tests/Microsoft.AspNetCore.Sockets.Client.Tests.csproj → test/Microsoft.AspNetCore.Sockets.Client.Tests/Microsoft.AspNetCore.Client.Tests.csproj
View File

@ -14,8 +14,8 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\Microsoft.AspNetCore.SignalR.Client\Microsoft.AspNetCore.SignalR.Client.csproj" />
<ProjectReference Include="..\..\src\Microsoft.AspNetCore.Sockets.Client\Microsoft.AspNetCore.Sockets.Client.csproj" />
<ProjectReference Include="..\..\src\Microsoft.AspNetCore.Sockets.Common\Microsoft.AspNetCore.Sockets.Common.csproj" />
<PackageReference Include="Microsoft.Extensions.Logging" Version="1.2.0-*" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.0.0-*" />
<PackageReference Include="Moq" Version="4.6.36-*" />