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

Hub<T> (#689)

This commit is contained in:
Mikael Mengistu 2017-08-08 17:39:09 -07:00 committed by GitHub
parent 3a1d4c5dd6
commit 32ef3eb355
10 changed files with 450 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
build/dependencies.props
View File

@ -16,6 +16,7 @@
<XunitVersion>2.3.0-beta2-*</XunitVersion>
<RxVersion>3.1.1</RxVersion>
<MsgPackVersion>0.9.0-beta2</MsgPackVersion>
<SystemReflectionEmitVersion>4.3.0</SystemReflectionEmitVersion>
<!--
TODO remove in next update of xunit
Prevent bug in xunit.analyzer from failing the build.

56
samples/SocketsSample/Hubs/HubTChat.cs Normal file
View File

@ -0,0 +1,56 @@
// 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.SignalR;
namespace SocketsSample.Hubs
{
public class HubTChat : Hub<IChatClient>
{
public override async Task OnConnectedAsync()
{
await Clients.All.Send($"{Context.ConnectionId} joined");
}
public override async Task OnDisconnectedAsync(Exception ex)
{
await Clients.All.Send($"{Context.ConnectionId} left");
}
public Task Send(string message)
{
return Clients.All.Send($"{Context.ConnectionId}: {message}");
}
public Task SendToGroup(string groupName, string message)
{
return Clients.Group(groupName).Send($"{Context.ConnectionId}@{groupName}: {message}");
}
public async Task JoinGroup(string groupName)
{
await Groups.AddAsync(Context.ConnectionId, groupName);
await Clients.Group(groupName).Send($"{Context.ConnectionId} joined {groupName}");
}
public async Task LeaveGroup(string groupName)
{
await Groups.RemoveAsync(Context.ConnectionId, groupName);
await Clients.Group(groupName).Send($"{Context.ConnectionId} left {groupName}");
}
public Task Echo(string message)
{
return Clients.Client(Context.ConnectionId).Send($"{Context.ConnectionId}: {message}");
}
}
public interface IChatClient
{
Task Send(string message);
}
}

1
samples/SocketsSample/Startup.cs
View File

@ -50,6 +50,7 @@ namespace SocketsSample
routes.MapHub<DynamicChat>("dynamic");
routes.MapHub<Chat>("default");
routes.MapHub<Streaming>("streaming");
routes.MapHub<HubTChat>("hubT");
});
app.UseSockets(routes =>

6
samples/SocketsSample/wwwroot/index.html
View File

@ -24,6 +24,12 @@
<li><a href="hubs.html?transport=ServerSentEvents&hubType=dynamic">Server Sent Events</a></li>
<li><a href="hubs.html?transport=WebSockets&hubType=dynamic">Web Sockets</a></li>
</ul>
<h1>ASP.NET Core SignalR (Hub&lt;T&gt;)</h1>
<ul>
<li><a href="hubs.html?transport=LongPolling&hubType=hubT">Long polling</a></li>
<li><a href="hubs.html?transport=ServerSentEvents&hubType=hubT">Server Sent Events</a></li>
<li><a href="hubs.html?transport=WebSockets&hubType=hubT">Web Sockets</a></li>
</ul>
<h1>ASP.NET Core SignalR (Streaming)</h1>
<ul>
<li><a href="streaming.html?transport=LongPolling">Long polling</a></li>

23
src/Microsoft.AspNetCore.SignalR/Hub`T.cs Normal file
View File

@ -0,0 +1,23 @@
// 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.
namespace Microsoft.AspNetCore.SignalR
{
public class Hub<T> : Hub where T : class
{
private IHubClients<T> _clients;
public new IHubClients<T> Clients
{
get
{
if (_clients == null)
{
_clients = new TypedHubClients<T>(base.Clients);
}
return _clients;
}
set { _clients = value; }
}
}
}

20
src/Microsoft.AspNetCore.SignalR/IHubClients`T.cs Normal file
View File

@ -0,0 +1,20 @@
// 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.Collections.Generic;
using System.Text;
namespace Microsoft.AspNetCore.SignalR
{
public interface IHubClients<T>
{
T All { get; }
T Client(string connectionId);
T Group(string groupName);
T User(string userId);
}
}

1
src/Microsoft.AspNetCore.SignalR/Microsoft.AspNetCore.SignalR.csproj
View File

@ -19,6 +19,7 @@
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="$(AspNetCoreVersion)" />
<PackageReference Include="Microsoft.Extensions.ClosedGenericMatcher.Sources" Version="$(AspNetCoreVersion)" PrivateAssets="All" />
<PackageReference Include="Microsoft.Extensions.ObjectMethodExecutor.Sources" Version="$(AspNetCoreVersion)" PrivateAssets="All" />
<PackageReference Include="System.Reflection.Emit" Version="$(SystemReflectionEmitVersion)" />
<PackageReference Include="Newtonsoft.Json" Version="$(JsonNetVersion)" />
</ItemGroup>

227
src/Microsoft.AspNetCore.SignalR/TypedClientBuilder.cs Normal file
View File

@ -0,0 +1,227 @@
// 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.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Reflection;
using System.Reflection.Emit;
using System.Threading.Tasks;
namespace Microsoft.AspNetCore.SignalR
{
internal static class TypedClientBuilder<T>
{
private const string ClientModuleName = "Microsoft.AspNetCore.SignalR.TypedClientBuilder";
// There is one static instance of _builder per T
private static Lazy<Func<IClientProxy, T>> _builder = new Lazy<Func<IClientProxy, T>>(() => GenerateClientBuilder());
public static T Build(IClientProxy proxy)
{
return _builder.Value(proxy);
}
public static void Validate()
{
// The following will throw if T is not a valid type
_ = _builder.Value;
}
private static Func<IClientProxy, T> GenerateClientBuilder()
{
VerifyInterface(typeof(T));
var assemblyName = new AssemblyName(ClientModuleName);
var assemblyBuilder = AssemblyBuilder.DefineDynamicAssembly(assemblyName, AssemblyBuilderAccess.Run);
var moduleBuilder = assemblyBuilder.DefineDynamicModule(ClientModuleName);
var clientType = GenerateInterfaceImplementation(moduleBuilder);
return proxy => (T)Activator.CreateInstance(clientType, proxy);
}
private static Type GenerateInterfaceImplementation(ModuleBuilder moduleBuilder)
{
var type = moduleBuilder.DefineType(
ClientModuleName + "." + typeof(T).Name + "Impl",
TypeAttributes.Public,
typeof(Object),
new[] { typeof(T) });
var proxyField = type.DefineField("_proxy", typeof(IClientProxy), FieldAttributes.Private);
BuildConstructor(type, proxyField);
foreach (var method in GetAllInterfaceMethods(typeof(T)))
{
BuildMethod(type, method, proxyField);
}
return type.CreateTypeInfo();
}
private static IEnumerable<MethodInfo> GetAllInterfaceMethods(Type interfaceType)
{
foreach (var parent in interfaceType.GetInterfaces())
{
foreach (var parentMethod in GetAllInterfaceMethods(parent))
{
yield return parentMethod;
}
}
foreach (var method in interfaceType.GetMethods())
{
yield return method;
}
}
private static void BuildConstructor(TypeBuilder type, FieldInfo proxyField)
{
var method = type.DefineMethod(".ctor", System.Reflection.MethodAttributes.Public | System.Reflection.MethodAttributes.HideBySig);
var ctor = typeof(object).GetConstructor(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic,
null, new Type[] { }, null);
method.SetReturnType(typeof(void));
method.SetParameters(typeof(IClientProxy));
var generator = method.GetILGenerator();
// Call object constructor
generator.Emit(OpCodes.Ldarg_0);
generator.Emit(OpCodes.Call, ctor);
// Assign constructor argument to the proxyField
generator.Emit(OpCodes.Ldarg_0); // type
generator.Emit(OpCodes.Ldarg_1); // type proxyfield
generator.Emit(OpCodes.Stfld, proxyField); // type.proxyField = proxyField
generator.Emit(OpCodes.Ret);
}
private static void BuildMethod(TypeBuilder type, MethodInfo interfaceMethodInfo, FieldInfo proxyField)
{
var methodAttributes =
MethodAttributes.Public
| MethodAttributes.Virtual
| MethodAttributes.Final
| MethodAttributes.HideBySig
| MethodAttributes.NewSlot;
ParameterInfo[] parameters = interfaceMethodInfo.GetParameters();
Type[] paramTypes = parameters.Select(param => param.ParameterType).ToArray();
var methodBuilder = type.DefineMethod(interfaceMethodInfo.Name, methodAttributes);
var invokeMethod = typeof(IClientProxy).GetMethod(
"InvokeAsync", BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, null,
new Type[] { typeof(string), typeof(object[]) }, null);
methodBuilder.SetReturnType(interfaceMethodInfo.ReturnType);
methodBuilder.SetParameters(paramTypes);
// Sets the number of generic type parameters
var genericTypeNames =
paramTypes.Where(p => p.IsGenericParameter).Select(p => p.Name).Distinct().ToArray();
if (genericTypeNames.Any())
{
methodBuilder.DefineGenericParameters(genericTypeNames);
}
var generator = methodBuilder.GetILGenerator();
// Declare local variable to store the arguments to IClientProxy.InvokeAsync
generator.DeclareLocal(typeof(object[]));
// Get IClientProxy
generator.Emit(OpCodes.Ldarg_0);
generator.Emit(OpCodes.Ldfld, proxyField);
// The first argument to IClientProxy.InvokeAsync is this method's name
generator.Emit(OpCodes.Ldstr, interfaceMethodInfo.Name);
// Create an new object array to hold all the parameters to this method
generator.Emit(OpCodes.Ldc_I4, parameters.Length); // Stack:
generator.Emit(OpCodes.Newarr, typeof(object)); // allocate object array
generator.Emit(OpCodes.Stloc_0);
// Store each parameter in the object array
for (int i = 0; i < paramTypes.Length; i++)
{
generator.Emit(OpCodes.Ldloc_0); // Object array loaded
generator.Emit(OpCodes.Ldc_I4, i);
generator.Emit(OpCodes.Ldarg, i + 1); // i + 1
generator.Emit(OpCodes.Box, paramTypes[i]);
generator.Emit(OpCodes.Stelem_Ref);
}
// Call InvokeAsync
generator.Emit(OpCodes.Ldloc_0);
generator.Emit(OpCodes.Callvirt, invokeMethod);
if (interfaceMethodInfo.ReturnType == typeof(void))
{
// void return
generator.Emit(OpCodes.Pop);
}
generator.Emit(OpCodes.Ret); // Return
}
private static void VerifyInterface(Type interfaceType)
{
if (!interfaceType.IsInterface)
{
throw new InvalidOperationException("Type must be an interface.");
}
if (interfaceType.GetProperties().Length != 0)
{
throw new InvalidOperationException("Type must not contain properties.");
}
if (interfaceType.GetEvents().Length != 0)
{
throw new InvalidOperationException("Type can not contain events.");
}
foreach (var method in interfaceType.GetMethods())
{
VerifyMethod(interfaceType, method);
}
foreach (var parent in interfaceType.GetInterfaces())
{
VerifyInterface(parent);
}
}
private static void VerifyMethod(Type interfaceType, MethodInfo interfaceMethod)
{
if (interfaceMethod.ReturnType != typeof(void) && interfaceMethod.ReturnType != typeof(Task))
{
throw new InvalidOperationException("Method must return Void or Task.");
}
foreach (var parameter in interfaceMethod.GetParameters())
{
VerifyParameter(interfaceType, interfaceMethod, parameter);
}
}
private static void VerifyParameter(Type interfaceType, MethodInfo interfaceMethod, ParameterInfo parameter)
{
if (parameter.IsOut)
{
throw new InvalidOperationException("Method must not take out parameters.");
}
if (parameter.ParameterType.IsByRef)
{
throw new InvalidOperationException("Method must not take reference parameters.");
}
}
}
}

32
src/Microsoft.AspNetCore.SignalR/TypedHubClients.cs Normal file
View File

@ -0,0 +1,32 @@
// 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.
namespace Microsoft.AspNetCore.SignalR
{
internal class TypedHubClients<T> : IHubClients<T>
{
private IHubClients hubClients;
public TypedHubClients(IHubClients dynamicContext)
{
hubClients = dynamicContext;
}
public T All => TypedClientBuilder<T>.Build(hubClients.All);
public T Client(string connectionId)
{
return TypedClientBuilder<T>.Build(hubClients.Client(connectionId));
}
public T Group(string groupName)
{
return TypedClientBuilder<T>.Build(hubClients.Group(groupName));
}
public T User(string userId)
{
return TypedClientBuilder<T>.Build(hubClients.User(userId));
}
}
}

83
test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTests.cs
View File

@ -605,6 +605,38 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
}
[Fact]
public async Task DelayedSendTest()
{
var serviceProvider = CreateServiceProvider();
dynamic endPoint = serviceProvider.GetService(GetEndPointType(typeof(HubT)));
using (var firstClient = new TestClient())
using (var secondClient = new TestClient())
{
Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
await firstClient.SendInvocationAsync("DelayedSend", secondClient.Connection.ConnectionId, "test").OrTimeout();
// check that 'secondConnection' has received the group send
var hubMessage = await secondClient.Read().OrTimeout();
var invocation = Assert.IsType<InvocationMessage>(hubMessage);
Assert.Equal("Send", invocation.Target);
Assert.Equal(1, invocation.Arguments.Length);
Assert.Equal("test", invocation.Arguments[0]);
// kill the connections
firstClient.Dispose();
secondClient.Dispose();
await Task.WhenAll(firstEndPointTask, secondEndPointTask).OrTimeout();
}
}
[Theory]
[InlineData(nameof(StreamingHub.CounterChannel))]
[InlineData(nameof(StreamingHub.CounterObservable))]
@ -732,6 +764,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
{
yield return new Type[] { typeof(DynamicTestHub) };
yield return new Type[] { typeof(MethodHub) };
yield return new Type[] { typeof(HubT) };
}
private static Type GetEndPointType(Type hubType)
@ -797,6 +830,56 @@ namespace Microsoft.AspNetCore.SignalR.Tests
}
}
public interface Test
{
Task Send(string message);
Task Broadcast(string message);
}
public class HubT : Hub<Test>
{
public override Task OnConnectedAsync()
{
var tcs = (TaskCompletionSource<bool>)Context.Connection.Metadata["ConnectedTask"];
tcs?.TrySetResult(true);
return base.OnConnectedAsync();
}
public string Echo(string data)
{
return data;
}
public Task ClientSendMethod(string userId, string message)
{
return Clients.User(userId).Send(message);
}
public Task ConnectionSendMethod(string connectionId, string message)
{
return Clients.Client(connectionId).Send(message);
}
public Task DelayedSend(string connectionId, string message)
{
Task.Delay(100);
return Clients.Client(connectionId).Send(message);
}
public Task GroupAddMethod(string groupName)
{
return Groups.AddAsync(Context.ConnectionId, groupName);
}
public Task GroupSendMethod(string groupName, string message)
{
return Clients.Group(groupName).Send(message);
}
public Task BroadcastMethod(string message)
{
return Clients.All.Broadcast(message);
}
}
public class StreamingHub : TestHub
{
public IObservable<string> CounterObservable(int count)