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

Redis presence manager

This commit is contained in:
Pawel Kadluczka 2017-04-27 10:01:12 -07:00
parent 7713124bd8
commit cf2d6ce697
7 changed files with 68 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
samples/ChatSample/HubWithPresence.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 System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR;
@ -24,12 +23,9 @@ namespace ChatSample
_userTracker = userTracker;
}
public Task<IEnumerable<UserDetails>> UsersOnline
public Task<IEnumerable<UserDetails>> GetUsersOnline()
{
get
{
return _userTracker.UsersOnline();
}
return _userTracker.UsersOnline();
}
public virtual Task OnUserJoined(UserDetails user)

3
samples/ChatSample/Hubs/Chat.cs
View File

@ -20,9 +20,10 @@ namespace ChatSample.Hubs
if (!Context.User.Identity.IsAuthenticated)
{
Context.Connection.Dispose();
return;
}
await Clients.Client(Context.ConnectionId).InvokeAsync("SetUsersOnline", await UsersOnline);
await Clients.Client(Context.ConnectionId).InvokeAsync("SetUsersOnline", await GetUsersOnline());
await base.OnConnectedAsync();
}

4
samples/ChatSample/IUserTracker.cs
View File

@ -11,8 +11,8 @@ namespace ChatSample
public interface IUserTracker<out THub>
{
Task<IEnumerable<UserDetails>> UsersOnline();
Task AddUser(Connection connection, UserDetails user);
Task<UserDetails> RemoveUser(Connection connection);
Task AddUser(Connection connection, UserDetails userDetails);
Task RemoveUser(Connection connection);
event Action<UserDetails> UserJoined;
event Action<UserDetails> UserLeft;

10
samples/ChatSample/InMemoryUserTracker.cs
View File

@ -18,22 +18,22 @@ namespace ChatSample
public Task<IEnumerable<UserDetails>> UsersOnline()
=> Task.FromResult(_usersOnline.Values.AsEnumerable());
public Task AddUser(Connection connection, UserDetails user)
public Task AddUser(Connection connection, UserDetails userDetails)
{
_usersOnline.TryAdd(connection, user);
UserJoined(user);
_usersOnline.TryAdd(connection, userDetails);
UserJoined(userDetails);
return Task.CompletedTask;
}
public Task<UserDetails> RemoveUser(Connection connection)
public Task RemoveUser(Connection connection)
{
if (_usersOnline.TryRemove(connection, out var userDetails))
{
UserLeft(userDetails);
}
return Task.FromResult(userDetails);
return Task.CompletedTask;
}
}
}

15
samples/ChatSample/PresenceHubLifetimeManager.cs
View File

@ -1,10 +1,13 @@

// 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;
using Microsoft.AspNetCore.Sockets;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.AspNetCore.SignalR.Redis;
namespace ChatSample
{
@ -18,6 +21,16 @@ namespace ChatSample
}
}
public class RedisPresenceHublifetimeMenager<THub> : PresenceHubLifetimeManager<THub, RedisHubLifetimeManager<THub>>
where THub : HubWithPresence
{
public RedisPresenceHublifetimeMenager(IUserTracker<THub> userTracker, IServiceScopeFactory serviceScopeFactory,
ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
: base(userTracker, serviceScopeFactory, loggerFactory, serviceProvider)
{
}
}
public class PresenceHubLifetimeManager<THub, THubLifetimeManager> : HubLifetimeManager<THub>, IDisposable
where THubLifetimeManager : HubLifetimeManager<THub>
where THub : HubWithPresence

117
samples/ChatSample/RedisPresenceManager.cs → samples/ChatSample/RedisUserTracker.cs
View File

@ -1,53 +1,37 @@
// 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.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR;
using Microsoft.AspNetCore.SignalR.Redis;
using Microsoft.AspNetCore.Sockets;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;
namespace ChatSample
{
public class RedisPresenceManager<THub> : IDisposable, IUserTracker<THub> where THub : HubWithPresence
public class RedisUserTracker<THub> : IUserTracker<THub>
{
private readonly RedisKey UsersOnlineRedisKey = "UsersOnline";
private readonly RedisChannel _redisChannel;
private IHubContext<THub> _hubContext;
private HubLifetimeManager<THub> _lifetimeManager;
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly int _redisDatabase;
private readonly ConnectionMultiplexer _redisConnection;
private readonly ISubscriber _redisSubscriber;
private readonly ILogger _logger;
private readonly RedisChannel _redisChannel;
private readonly ConcurrentDictionary<Connection, object> _connections
= new ConcurrentDictionary<Connection, object>();
public event Action<UserDetails> UserJoined;
public event Action<UserDetails> UserLeft;
// TODO: subscribe and handle lifecycle events/disconnects
// TODO: handle situations where Redis shuts down
// TODO: handle situations where a server goes down and the server has zombie connections
public RedisPresenceManager(IHubContext<THub> hubContext, HubLifetimeManager<THub> lifetimeManager,
IServiceScopeFactory serviceScopeFactory, IOptions<RedisOptions> options, ILoggerFactory loggerFactory)
public RedisUserTracker(IOptions<RedisOptions> options, ILoggerFactory loggerFactory)
{
_hubContext = hubContext;
_lifetimeManager = lifetimeManager;
_serviceScopeFactory = serviceScopeFactory;
_logger = loggerFactory.CreateLogger<RedisPresenceManager<THub>>();
_logger = loggerFactory.CreateLogger<RedisUserTracker<THub>>();
_redisDatabase = options.Value.Options.DefaultDatabase.GetValueOrDefault();
_redisConnection = ConnectToRedis(options.Value, _logger);
_redisSubscriber = _redisConnection.GetSubscriber();
@ -59,11 +43,11 @@ namespace ChatSample
var user = ToUserDetails(stringValue.Substring(1));
if (stringValue[0] == '-')
{
_ = Notify(hub => hub.OnUserLeft(user));
UserLeft(user);
}
else
{
_ = Notify(hub => hub.OnUserJoined(user));
UserJoined(user);
}
});
}
@ -71,17 +55,20 @@ namespace ChatSample
private static ConnectionMultiplexer ConnectToRedis(RedisOptions options, ILogger logger)
{
var loggerTextWriter = new LoggerTextWriter(logger);
return options.Factory == null
? ConnectionMultiplexer.Connect(options.Options, loggerTextWriter)
: options.Factory(loggerTextWriter);
}
if (options.Factory != null)
{
return options.Factory(loggerTextWriter);
}
public async Task<IEnumerable<UserDetails>> UsersOnline()
{
var database = _redisConnection.GetDatabase(_redisDatabase);
var usersOnline = await database.SetMembersAsync(UsersOnlineRedisKey);
if (options.Options.EndPoints.Any())
{
return ConnectionMultiplexer.Connect(options.Options, loggerTextWriter);
}
return usersOnline.Select(u => ToUserDetails(u));
var configurationOptions = new ConfigurationOptions();
configurationOptions.EndPoints.Add(IPAddress.Loopback, 0);
configurationOptions.SetDefaultPorts();
return ConnectionMultiplexer.Connect(configurationOptions, loggerTextWriter);
}
private static UserDetails ToUserDetails(string user)
@ -91,66 +78,30 @@ namespace ChatSample
return new UserDetails(user.Substring(0, pos), user.Substring(pos + 1));
}
public Task UserJoined(Connection connection)
public async Task AddUser(Connection connection, UserDetails userDetails)
{
_connections.TryAdd(connection, null);
var database = _redisConnection.GetDatabase(_redisDatabase);
var user = $"{connection.ConnectionId}|{connection.User.Identity.Name}";
// Fire and forget
_ = database.SetAddAsync(UsersOnlineRedisKey, $"{connection.ConnectionId}|{connection.User.Identity.Name}");
// need to await to make sure user is added before we call into the Hub
await database.SetAddAsync(UsersOnlineRedisKey, $"{connection.ConnectionId}|{connection.User.Identity.Name}");
_ = _redisSubscriber.PublishAsync(_redisChannel, "+" + user);
return Task.CompletedTask;
}
public Task UserLeft(Connection connection)
public async Task RemoveUser(Connection connection)
{
_connections.TryRemove(connection, out object _);
var database = _redisConnection.GetDatabase(_redisDatabase);
var user = $"{connection.ConnectionId}|{connection.User.Identity.Name}";
// Fire and forget
_ = database.SetRemoveAsync(UsersOnlineRedisKey, user);
await database.SetRemoveAsync(UsersOnlineRedisKey, user);
_ = _redisSubscriber.PublishAsync(_redisChannel, "-" + user);
return Task.CompletedTask;
}
private async Task Notify(Func<THub, Task> invocation)
public async Task<IEnumerable<UserDetails>> UsersOnline()
{
foreach (var connection in _connections.Keys)
{
using (var scope = _serviceScopeFactory.CreateScope())
{
var hubActivator = scope.ServiceProvider.GetRequiredService<IHubActivator<THub, IClientProxy>>();
var hub = hubActivator.Create();
hub.Clients = _hubContext.Clients;
hub.Context = new HubCallerContext(connection);
hub.Groups = new GroupManager<THub>(connection, _lifetimeManager);
try
{
await invocation(hub);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Presence notification failed.");
}
finally
{
hubActivator.Release(hub);
}
}
}
}
public void Dispose()
{
_redisSubscriber.Unsubscribe(_redisChannel);
_redisConnection.Close();
_redisConnection.Dispose();
var database = _redisConnection.GetDatabase(_redisDatabase);
var usersOnline = await database.SetMembersAsync(UsersOnlineRedisKey);
return usersOnline.Select(u => ToUserDetails(u));
}
private class LoggerTextWriter : TextWriter
@ -171,8 +122,8 @@ namespace ChatSample
public override void WriteLine(string value)
{
_logger.LogDebug(value);
}
}
}
}
*/
}

10
samples/ChatSample/Startup.cs
View File

@ -9,6 +9,7 @@ using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Identity.EntityFrameworkCore;
using Microsoft.AspNetCore.SignalR;
using Microsoft.AspNetCore.SignalR.Redis;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
@ -54,15 +55,22 @@ namespace ChatSample
services.AddTransient<IEmailSender, AuthMessageSender>();
services.AddTransient<ISmsSender, AuthMessageSender>();
// To use Redis scaleout uncomment .AddRedis
// To use Redis scaleout uncomment .AddRedis and uncomment Redis related lines below for presence
services.AddSignalR()
// .AddRedis()
;
services.AddAuthentication();
services.AddSingleton(typeof(DefaultHubLifetimeManager<>), typeof(DefaultHubLifetimeManager<>));
services.AddSingleton(typeof(HubLifetimeManager<>), typeof(DefaultPresenceHublifetimeMenager<>));
services.AddSingleton(typeof(IUserTracker<>), typeof(InMemoryUserTracker<>));
/*
services.AddSingleton(typeof(RedisHubLifetimeManager<>), typeof(RedisHubLifetimeManager<>));
services.AddSingleton(typeof(HubLifetimeManager<>), typeof(RedisPresenceHublifetimeMenager<>));
services.AddSingleton(typeof(IUserTracker<>), typeof(RedisUserTracker<>));
*/
}
// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.