增加会话服务及统一下发服务的实现

src/JT808.DotNetty.Hosting/Program.cs

@@ -36,6 +36,7 @@ namespace JT808.DotNetty.Hosting
                     services.Replace(new ServiceDescriptor(typeof(JT808MsgIdHandlerBase), typeof(JT808MsgIdCustomHandler), ServiceLifetime.Singleton));
                 })
                 .UseJT808Host();
+
             await serverHostBuilder.RunConsoleAsync();
         }
     }

src/JT808.DotNetty/Dtos/JT808ResultDto.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Dtos
+{
+    public class JT808ResultDto<T>
+    {
+        public string Message { get; set; }
+
+        public int Code { get; set; }
+
+        public T Data { get; set; }
+    }
+}

src/JT808.DotNetty/Dtos/JT808SessionInfoDto.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Dtos
+{
+    public class JT808SessionInfoDto
+    {
+        /// <summary>
+        /// 通道Id
+        /// </summary>
+        public string ChannelId { get; set; }
+        /// <summary>
+        /// 最后上线时间
+        /// </summary>
+        public DateTime LastActiveTime { get; set; }
+        /// <summary>
+        /// 上线时间
+        /// </summary>
+        public DateTime StartTime { get; set; }
+        /// <summary>
+        /// 终端手机号
+        /// </summary>
+        public string TerminalPhoneNo { get; set; }
+    }
+}

src/JT808.DotNetty/Handlers/JT808ConnectionHandler.cs

@@ -1,5 +1,6 @@
 using DotNetty.Handlers.Timeout;
 using DotNetty.Transport.Channels;
+using JT808.DotNetty.Metadata;
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
@@ -32,6 +33,7 @@ namespace JT808.DotNetty.Handlers
             string channelId = context.Channel.Id.AsShortText();
             if (logger.IsEnabled(LogLevel.Debug))
                 logger.LogDebug($"<<<{ channelId } Successful client connection to server.");
+            jT808SessionManager.TryAddSession(new JT808Session(context.Channel));
             base.ChannelActive(context);
         }
 

src/JT808.DotNetty/Interfaces/IJT808SessionService.cs

@@ -0,0 +1,58 @@
+using JT808.DotNetty.Dtos;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Interfaces
+{
+    /// <summary>
+    /// JT808会话服务
+    /// </summary>
+    public interface IJT808SessionService
+    {
+        /// <summary>
+        /// 获取真实连接数
+        /// </summary>
+        /// <returns></returns>
+        JT808ResultDto<int> GetRealLinkCount();
+        /// <summary>
+        /// 获取设备相关连的连接数
+        /// </summary>
+        /// <returns></returns>
+        JT808ResultDto<int> GetRelevanceLinkCount();
+        /// <summary>
+        /// 获取实际会话集合
+        /// </summary>
+        /// <returns></returns>
+        JT808ResultDto<List<JT808SessionInfoDto>> GetRealAll();
+        /// <summary>
+        /// 获取设备相关连会话集合
+        /// </summary>
+        /// <returns></returns>
+        JT808ResultDto<List<JT808SessionInfoDto>> GetRelevanceAll();
+        /// <summary>
+        /// 通过通道Id移除对应会话
+        /// </summary>
+        /// <param name="channelId"></param>
+        /// <returns></returns>
+        JT808ResultDto<bool> RemoveByChannelId(string channelId);
+        /// <summary>
+        /// 通过设备终端号移除对应会话
+        /// </summary>
+        /// <param name="terminalPhoneNo"></param>
+        /// <returns></returns>
+        JT808ResultDto<bool> RemoveByTerminalPhoneNo(string terminalPhoneNo);
+        /// <summary>
+        /// 通过通道Id获取会话信息
+        /// </summary>
+        /// <param name="channelId"></param>
+        /// <returns></returns>
+        JT808ResultDto<JT808SessionInfoDto> GetByChannelId(string channelId);
+        /// <summary>
+        /// 通过设备终端号获取会话信息
+        /// </summary>
+        /// <param name="terminalPhoneNo"></param>
+        /// <returns></returns>
+        JT808ResultDto<JT808SessionInfoDto> GetByTerminalPhoneNo(string terminalPhoneNo);
+    }
+}

src/JT808.DotNetty/Interfaces/IJT808UnificationSendService.cs

@@ -0,0 +1,15 @@
+using JT808.DotNetty.Dtos;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Interfaces
+{
+    /// <summary>
+    /// JT808统一下发命令
+    /// </summary>
+    public interface IJT808UnificationSendService
+    {
+        JT808ResultDto<bool> Send(string terminalPhoneNo, byte[] data);
+    }
+}

src/JT808.DotNetty/Internal/JT808SessionServiceDefaultImpl.cs

@@ -0,0 +1,203 @@
+using JT808.DotNetty.Dtos;
+using JT808.DotNetty.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace JT808.DotNetty.Internal
+{
+    internal class JT808SessionServiceDefaultImpl : IJT808SessionService
+    {
+        private readonly JT808SessionManager jT808SessionManager;
+
+        public JT808SessionServiceDefaultImpl(JT808SessionManager jT808SessionManager)
+        {
+            this.jT808SessionManager = jT808SessionManager;
+        }
+
+        public JT808ResultDto<JT808SessionInfoDto> GetByChannelId(string channelId)
+        {
+            JT808ResultDto<JT808SessionInfoDto> resultDto = new JT808ResultDto<JT808SessionInfoDto>();
+            try
+            {
+                var result = jT808SessionManager.GetSessionByID(channelId);
+                JT808SessionInfoDto jT808SessionInfoDto = new JT808SessionInfoDto
+                {
+                    TerminalPhoneNo = result.TerminalPhoneNo,
+                    ChannelId=result.SessionID,
+                    LastActiveTime=result.LastActiveTime,
+                    StartTime=result.StartTime
+                };
+                resultDto.Data = jT808SessionInfoDto;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = null;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<int> GetRelevanceLinkCount()
+        {
+            JT808ResultDto<int> resultDto = new JT808ResultDto<int>();
+            try
+            {
+                resultDto.Data = jT808SessionManager.RelevanceSessionCount;
+                resultDto.Code = 200;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = 0;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<int> GetRealLinkCount()
+        {
+            JT808ResultDto<int> resultDto = new JT808ResultDto<int>();
+            try
+            {
+                resultDto.Data = jT808SessionManager.RealSessionCount;
+                resultDto.Code = 200;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = 0;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<List<JT808SessionInfoDto>> GetRealAll()
+        {
+            JT808ResultDto<List<JT808SessionInfoDto>> resultDto = new JT808ResultDto<List<JT808SessionInfoDto>>();
+            try
+            {
+                resultDto.Data = jT808SessionManager.GetRealAll().Select(s => new JT808SessionInfoDto
+                {
+                    ChannelId = s.SessionID,
+                    LastActiveTime = s.LastActiveTime,
+                    StartTime = s.StartTime,
+                    TerminalPhoneNo = s.TerminalPhoneNo
+                }).ToList();
+                resultDto.Code = 200;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = null;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<List<JT808SessionInfoDto>> GetRelevanceAll()
+        {
+            JT808ResultDto<List<JT808SessionInfoDto>> resultDto = new JT808ResultDto<List<JT808SessionInfoDto>>();
+            try
+            {
+                resultDto.Data = jT808SessionManager.GetRelevanceAll().Select(s => new JT808SessionInfoDto
+                {
+                    ChannelId = s.SessionID,
+                    LastActiveTime = s.LastActiveTime,
+                    StartTime = s.StartTime,
+                    TerminalPhoneNo = s.TerminalPhoneNo
+                }).ToList();
+                resultDto.Code = 200;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = null;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<bool> RemoveByChannelId(string channelId)
+        {
+            JT808ResultDto<bool> resultDto = new JT808ResultDto<bool>();
+            try
+            {
+                var session = jT808SessionManager.RemoveSessionByID(channelId);
+                if (session != null)
+                {
+                    session.Channel.CloseAsync();
+                }
+                resultDto.Code = 200;
+                resultDto.Data = true;
+            }
+            catch (AggregateException ex)
+            {
+                resultDto.Data = false;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = false;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<bool> RemoveByTerminalPhoneNo(string terminalPhoneNo)
+        {
+            JT808ResultDto<bool> resultDto = new JT808ResultDto<bool>();
+            try
+            {
+                var session = jT808SessionManager.RemoveSessionByTerminalPhoneNo(terminalPhoneNo);
+                if (session != null)
+                {
+                    session.Channel.CloseAsync();
+                }
+                resultDto.Code = 200;
+                resultDto.Data = true;
+            }
+            catch (AggregateException ex)
+            {
+                resultDto.Data = false;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = false;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+
+        public JT808ResultDto<JT808SessionInfoDto> GetByTerminalPhoneNo(string terminalPhoneNo)
+        {
+            JT808ResultDto<JT808SessionInfoDto> resultDto = new JT808ResultDto<JT808SessionInfoDto>();
+            try
+            {
+                var result = jT808SessionManager.GetSessionByTerminalPhoneNo(terminalPhoneNo);
+                JT808SessionInfoDto jT808SessionInfoDto = new JT808SessionInfoDto
+                {
+                    TerminalPhoneNo = result.TerminalPhoneNo,
+                    ChannelId = result.SessionID,
+                    LastActiveTime = result.LastActiveTime,
+                    StartTime = result.StartTime
+                };
+                resultDto.Data = jT808SessionInfoDto;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = null;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+    }
+}

src/JT808.DotNetty/Internal/JT808UnificationSendServiceDefaultImpl.cs

@@ -0,0 +1,37 @@
+using DotNetty.Buffers;
+using JT808.DotNetty.Dtos;
+using JT808.DotNetty.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Internal
+{
+    internal class JT808UnificationSendServiceDefaultImpl : IJT808UnificationSendService
+    {
+        private readonly JT808SessionManager jT808SessionManager;
+
+        public JT808UnificationSendServiceDefaultImpl(JT808SessionManager jT808SessionManager)
+        {
+            this.jT808SessionManager = jT808SessionManager;
+        }
+
+        public JT808ResultDto<bool> Send(string terminalPhoneNo, byte[] data)
+        {
+            JT808ResultDto<bool> resultDto = new JT808ResultDto<bool>();
+            try
+            {
+                var session = jT808SessionManager.GetSessionByTerminalPhoneNo(terminalPhoneNo);
+                session.Channel.WriteAndFlushAsync(Unpooled.WrappedBuffer(data));
+                resultDto.Data = true;
+            }
+            catch (Exception ex)
+            {
+                resultDto.Data = false;
+                resultDto.Code = 500;
+                resultDto.Message = Newtonsoft.Json.JsonConvert.SerializeObject(ex);
+            }
+            return resultDto;
+        }
+    }
+}

src/JT808.DotNetty/JT808.DotNetty.csproj

@@ -14,6 +14,7 @@
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.1.1" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="2.1.1" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.1.1" />
+    <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
     <PackageReference Include="Polly" Version="6.1.0" />
   </ItemGroup>
 

src/JT808.DotNetty/JT808DotnettyExtensions.cs

@@ -16,16 +16,32 @@ namespace JT808.DotNetty
         public static IHostBuilder UseJT808Host(this IHostBuilder builder)
         {
             return builder.ConfigureServices((hostContext, services) =>
-                 {
-                     services.Configure<JT808Configuration>(hostContext.Configuration.GetSection("JT808Configuration"));
-                     services.TryAddSingleton<JT808SessionManager>();
-                     services.TryAddSingleton<JT808MsgIdHandlerBase,JT808MsgIdDefaultHandler>();
-                     services.TryAddSingleton<IJT808SourcePackageDispatcher, JT808SourcePackageDispatcherDefaultImpl>();
-                     services.TryAddScoped<JT808ConnectionHandler>();
-                     services.TryAddScoped<JT808Decoder>();
-                     services.TryAddScoped<JT808ServerHandler>();
-                     services.AddHostedService<JT808ServerHost>();
-                 });
+            {
+                services.Configure<JT808Configuration>(hostContext.Configuration.GetSection("JT808Configuration"));
+                services.TryAddSingleton<JT808SessionManager>();
+                services.TryAddSingleton<JT808MsgIdHandlerBase,JT808MsgIdDefaultHandler>();
+                services.TryAddSingleton<IJT808SourcePackageDispatcher, JT808SourcePackageDispatcherDefaultImpl>();
+                services.TryAddScoped<JT808ConnectionHandler>();
+                services.TryAddScoped<JT808Decoder>();
+                services.TryAddScoped<JT808ServerHandler>();
+                services.TryAddSingleton<IJT808SessionService, JT808SessionServiceDefaultImpl>();
+                services.TryAddSingleton<IJT808UnificationSendService, JT808UnificationSendServiceDefaultImpl>();
+                services.AddHostedService<JT808ServerHost>();
+            });
+        }
+
+        public static void UseJT808Host(this IServiceCollection  serviceDescriptors, HostBuilderContext hostContext)
+        {
+            serviceDescriptors.Configure<JT808Configuration>(hostContext.Configuration.GetSection("JT808Configuration"));
+            serviceDescriptors.TryAddSingleton<JT808SessionManager>();
+            serviceDescriptors.TryAddSingleton<JT808MsgIdHandlerBase, JT808MsgIdDefaultHandler>();
+            serviceDescriptors.TryAddSingleton<IJT808SourcePackageDispatcher, JT808SourcePackageDispatcherDefaultImpl>();
+            serviceDescriptors.TryAddScoped<JT808ConnectionHandler>();
+            serviceDescriptors.TryAddScoped<JT808Decoder>();
+            serviceDescriptors.TryAddScoped<JT808ServerHandler>();
+            serviceDescriptors.TryAddSingleton<IJT808SessionService, JT808SessionServiceDefaultImpl>();
+            serviceDescriptors.TryAddSingleton<IJT808UnificationSendService, JT808UnificationSendServiceDefaultImpl>();
+            serviceDescriptors.AddHostedService<JT808ServerHost>();
         }
     }
 }

src/JT808.DotNetty/JT808SessionManager.cs

@@ -29,8 +29,8 @@ namespace JT808.DotNetty
             {
                 while (!cancellationTokenSource.IsCancellationRequested)
                 {
-                    logger.LogInformation($"Online Count>>>{SessionCount}");
-                    if (SessionCount > 0)
+                    logger.LogInformation($"Online Count>>>{RealSessionCount}");
+                    if (RealSessionCount > 0)
                     {
                         logger.LogInformation($"SessionIds>>>{string.Join(",", SessionIdDict.Select(s => s.Key))}");
                         logger.LogInformation($"TerminalPhoneNos>>>{string.Join(",", TerminalPhoneNo_SessionId_Dict.Select(s => $"{s.Key}-{s.Value}"))}");
@@ -53,7 +53,10 @@ namespace JT808.DotNetty
         /// </summary>
         private ConcurrentDictionary<string, string> TerminalPhoneNo_SessionId_Dict = new ConcurrentDictionary<string, string>(StringComparer.OrdinalIgnoreCase);
 
-        public int SessionCount
+        /// <summary>
+        /// 实际连接数
+        /// </summary>
+        public int RealSessionCount
         {
             get
             {
@@ -61,6 +64,17 @@ namespace JT808.DotNetty
             }
         }
 
+        /// <summary>
+        /// 获取设备相关连的连接数
+        /// </summary>
+        public int RelevanceSessionCount
+        {
+            get
+            {
+                return TerminalPhoneNo_SessionId_Dict.Count;
+            }
+        }
+
         public JT808Session GetSessionByID(string sessionID)
         {
             if (string.IsNullOrEmpty(sessionID))
@@ -130,9 +144,14 @@ namespace JT808.DotNetty
             TerminalPhoneNo_SessionId_Dict.AddOrUpdate(appSession.TerminalPhoneNo, appSession.SessionID, (x, y) => appSession.SessionID);
         }
 
-        public void RemoveSessionByID(string sessionID)
+        public void TryAddSession(JT808Session appSession)
+        {
+            SessionIdDict.AddOrUpdate(appSession.SessionID, appSession, (x, y) => appSession);
+        }
+
+        public JT808Session RemoveSessionByID(string sessionID)
         {
-            if (sessionID == null) return;
+            if (sessionID == null) return null;
             try
             {
                 if (SessionIdDict.TryRemove(sessionID, out JT808Session session))
@@ -148,14 +167,52 @@ namespace JT808.DotNetty
                     {
                         logger.LogInformation($">>>{sessionID} Session Remove.");
                     }
+                    return session;
                 }
+                return null;
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, $">>>{sessionID} Session Remove Exception");
             }
+            return null;
         }
 
+        public JT808Session RemoveSessionByTerminalPhoneNo(string terminalPhoneNo)
+        {
+            if (terminalPhoneNo == null) return null;
+            try
+            {
+                if (TerminalPhoneNo_SessionId_Dict.TryRemove(terminalPhoneNo, out string sessionid))
+                {
+                    if (SessionIdDict.TryRemove(sessionid, out JT808Session session))
+                    {
+                        logger.LogInformation($">>>{sessionid}-{session.TerminalPhoneNo} Session Remove.");
+                        return session;
+                    }
+                    else
+                    {
+                        return null;
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, $">>>{terminalPhoneNo} Session Remove Exception");
+            }
+            return null;
+        }
+
+        public IEnumerable<JT808Session> GetRealAll()
+        {
+            return SessionIdDict.Select(s=>s.Value);
+        }
+
+        public List<JT808Session> GetRelevanceAll()
+        {
+            return SessionIdDict.Join(TerminalPhoneNo_SessionId_Dict, m => m.Key, s => s.Value, (m, s) => m.Value).ToList();
+        }
+       
         public void Dispose()
         {
             cancellationTokenSource.Cancel();

src/JT808.DotNetty/Metadata/JT808Session.cs

@@ -16,6 +16,14 @@ namespace JT808.DotNetty.Metadata
             SessionID = Channel.Id.AsShortText();
         }
 
+        public JT808Session(IChannel channel)
+        {
+            Channel = channel;
+            StartTime = DateTime.Now;
+            LastActiveTime = DateTime.Now;
+            SessionID = Channel.Id.AsShortText();
+        }
+
         /// <summary>
         /// 终端手机号
         /// </summary>