hotline/src/Hotline.Api/Realtimes/HotlineHub.cs

using Hotline.Caching.Interfaces;
using Hotline.Realtimes;
using Hotline.Share.Dtos.Realtime;
using Hotline.Users;
using Microsoft.AspNetCore.SignalR;
using XF.Domain.Authentications;
using XF.Domain.Cache;
using XF.Domain.Exceptions;

namespace Hotline.Api.Realtimes
{
    public class HotlineHub : Hub
    {
        private readonly ISessionContext _sessionContext;
        private readonly IWorkRepository _workRepository;
        private readonly ITypedCache<RealtimeConnection> _cacheConnection;
        private readonly IUserCacheManager _userCacheManager;

        public HotlineHub(
            ISessionContext sessionContext,
            IWorkRepository workRepository,
            ITypedCache<RealtimeConnection> cacheConnection,
            IUserCacheManager userCacheManager)
        {
            _sessionContext = sessionContext;
            _workRepository = workRepository;
            _cacheConnection = cacheConnection;
            _userCacheManager = userCacheManager;
        }

        /// <summary>
        /// Called when a new connection is established with the hub.
        /// </summary>
        /// <returns>A <see cref="T:System.Threading.Tasks.Task" /> that represents the asynchronous connect.</returns>
        public override async Task OnConnectedAsync()
        {
            var userId = _sessionContext.RequiredUserId;
            //todo 检查是否exists , t:说明可能是断线重连，1.set cache 2.依据缓存加入group f: 1.set cache
            //await SetConnectionAsync(new RealtimeConnection(userId, Context.ConnectionId), Context.ConnectionAborted);
            var connection = await _cacheConnection.GetAsync(userId, Context.ConnectionAborted);
            if (connection == null)
            {
                await SetConnectionAsync(new RealtimeConnection(userId, Context.ConnectionId), Context.ConnectionAborted);
            }
            else
            {
                connection.ConnectionId = Context.ConnectionId;
                await SetConnectionAsync(connection, Context.ConnectionAborted);

                if (connection.GroupNames.Any())
                {
                    foreach (var groupName in connection.GroupNames)
                    {
                        await Groups.AddToGroupAsync(Context.ConnectionId, groupName);
                        await Clients.Group(groupName).SendAsync("Send", $"{Context.ConnectionId} has joined the group {groupName}.");
                    }
                }
            }

            await base.OnConnectedAsync();
        }

        /// <summary>Called when a connection with the hub is terminated.</summary>
        /// <returns>A <see cref="T:System.Threading.Tasks.Task" /> that represents the asynchronous disconnect.</returns>
        public override async Task OnDisconnectedAsync(Exception? exception)
        {
            var userId = _sessionContext.RequiredUserId;
            await _cacheConnection.RemoveAsync(userId, default);
            await base.OnDisconnectedAsync(exception);
        }

        public async Task JoinGroupAsync(JoinGroupDto dto)
        {
            CheckIfConnected();

            var userId = _sessionContext.RequiredUserId;
            var connection = await GetConnectionAsync(userId, Context.ConnectionAborted);
            if (connection is null)
                throw new UserFriendlyException($"尚未建立实时服务连接, userId:{userId}", "尚未建立实时服务连接");

            connection.GroupNames.Add(dto.GroupName);
            connection.GroupNames = connection.GroupNames.Distinct().ToList();
            await SetConnectionAsync(connection, Context.ConnectionAborted);

            //switch (dto.GroupName)
            //{
            //    case RealtimeGroupNames.CallCenter:
            //        var work = _userCacheManager.GetWorkByUser(userId);
            //        if (work == null)
            //            throw new UserFriendlyException($"未查询到上班记录, userId: {userId}", "未查询到上班记录");
            //        work.SignalRId = Context.ConnectionId;
            //        await _workRepository.UpdateAsync(work, Context.ConnectionAborted);
            //        _userCacheManager.UpdateWorkByUser(work);
            //        break;
            //}

            await Groups.AddToGroupAsync(Context.ConnectionId, dto.GroupName);
            await Clients.Group(dto.GroupName).SendAsync("Send", $"{Context.ConnectionId} has joined the group {dto.GroupName}.");
        }

        public async Task LeaveGroupAsync(JoinGroupDto dto)
        {
            CheckIfConnected();

            var userId = _sessionContext.RequiredUserId;
            var connection = await GetConnectionAsync(userId, Context.ConnectionAborted);
            if (connection is null)
                throw new UserFriendlyException($"尚未建立实时服务连接, userId:{userId}", "尚未建立实时服务连接");

            if (connection.GroupNames.Contains(dto.GroupName))
            {
                connection.GroupNames.Remove(dto.GroupName);
                await SetConnectionAsync(connection, Context.ConnectionAborted);
            }

            await Groups.RemoveFromGroupAsync(Context.ConnectionId, dto.GroupName);
            await Clients.Group(dto.GroupName).SendAsync("Send", $"{Context.ConnectionId} has left the group {dto.GroupName}.");

            //check if not in any groups then disconnect the connection
            //if (!connection.GroupNames.Any())
            //    Context.Abort();
        }

        #region private

        private void CheckIfConnected()
        {
            if (string.IsNullOrEmpty(Context.ConnectionId))
                throw UserFriendlyException.SameMessage("尚未建立实时通信连接");
        }

        private Task<RealtimeConnection?> GetConnectionAsync(string userId, CancellationToken cancellationToken)
        {
            return _cacheConnection.GetAsync(userId, cancellationToken);
        }

        private Task SetConnectionAsync(RealtimeConnection connection, CancellationToken cancellationToken)
        {
            return _cacheConnection.SetAsync(connection.UserId, connection, TimeSpan.FromDays(7), cancellationToken);
        }
        #endregion
    }
}