1.增加生产消费模式测试

2.增加内部应答消费服务
3.去掉无用文件

simples/JT808.DotNetty.SimpleClient/Services/UpService.cs

@@ -20,7 +20,7 @@ namespace JT808.DotNetty.SimpleClient.Services
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            JT808TcpClient client1 = jT808TcpClientFactory.Create(new DeviceConfig("12345678910", "127.0.0.1", 12808));
+            JT808TcpClient client1 = jT808TcpClientFactory.Create(new DeviceConfig("12345678910", "127.0.0.1", 808));
             //1.终端注册
             client1.Send(new JT808_0x0100()
             {
@@ -41,7 +41,7 @@ namespace JT808.DotNetty.SimpleClient.Services
                 while (true)
                 {
                     var i = 0;
-                    //3.每30秒发一次
+                    //3.每5000秒发一次
                     client1.Send(new JT808_0x0200()
                     {
                         Lat = 110000 + i,
@@ -54,7 +54,7 @@ namespace JT808.DotNetty.SimpleClient.Services
                         StatusFlag = 10
                     });
                     i++;
-                    Thread.Sleep(30000);
+                    Thread.Sleep(5000);
                 }
             });
             return Task.CompletedTask;

simples/JT808.DotNetty.SimpleServer/Handlers/JT808MsgIdTcpSimpleHandler.cs

@@ -1,60 +0,0 @@
-using JT808.DotNetty.Core;
-using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Metadata;
-using JT808.DotNetty.Core.Session;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.SimpleServer.Handlers
-{
-    public class JT808MsgIdTcpSimpleHandler : JT808MsgIdTcpHandlerBase
-    {
-        public JT808MsgIdTcpSimpleHandler(
-            ILoggerFactory loggerFactory,
-            JT808SessionManager sessionManager) : base(sessionManager)
-        {
-            logger = loggerFactory.CreateLogger<JT808MsgIdTcpSimpleHandler>();
-        }
-
-        private readonly ILogger<JT808MsgIdTcpSimpleHandler> logger;
-
-        public override IJT808Reply Msg0x0200(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0200");
-            return base.Msg0x0200(request);
-        }
-
-        public override IJT808Reply Msg0x0001(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0001");
-            return base.Msg0x0001(request);
-        }
-
-        public override IJT808Reply Msg0x0002(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0002");
-            return base.Msg0x0002(request);
-        }
-
-        public override IJT808Reply Msg0x0003(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0003");
-            return base.Msg0x0003(request);
-        }
-
-        public override IJT808Reply Msg0x0100(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0100");
-            return base.Msg0x0100(request);
-        }
-
-        public override IJT808Reply Msg0x0102(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0102");
-            return base.Msg0x0102(request);
-        }
-    }
-}

simples/JT808.DotNetty.SimpleServer/Program.cs

@@ -1,7 +1,6 @@
 
 using JT808.DotNetty.Core;
 using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.SimpleServer.Handlers;
 using JT808.DotNetty.Tcp;
 using JT808.Protocol;
 using Microsoft.Extensions.Configuration;

src/JT808.DotNetty.Abstractions/IJT808DownlinkPacket.cs

@@ -1,22 +0,0 @@
-using JT808.DotNetty.Abstractions.Enums;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JT808.DotNetty.Abstractions
-{
-    /// <summary>
-    /// 下行数据包处理接口
-    /// </summary>
-    public interface IJT808DownlinkPacket
-    {
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="data">数据包</param>
-        /// <param name="protocolType">协议类型:tcp/udp</param>
-        /// <returns></returns>
-        Task ProcessorAsync(byte[] data, JT808TransportProtocolType transportProtocolType);
-    }
-}

src/JT808.DotNetty.Abstractions/IJT808MsgConsumer.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+
+namespace JT808.DotNetty.Abstractions
+{
+    public interface IJT808MsgConsumer : IJT808PubSub, IDisposable
+    {
+        void OnMessage(Action<(string TerminalNo, byte[] Data)> callback);
+        CancellationTokenSource Cts { get; }
+        void Subscribe();
+        void Unsubscribe();
+    }
+}

src/JT808.DotNetty.Abstractions/IJT808MsgProducer.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JT808.DotNetty.Abstractions
+{
+    public interface IJT808MsgProducer : IJT808PubSub, IDisposable
+    {
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="terminalNo">设备终端号</param>
+        /// <param name="data">808 hex data</param>
+        Task ProduceAsync(string terminalNo, byte[] data);
+    }
+}

src/JT808.DotNetty.Abstractions/IJT808MsgReplyConsumer.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+
+namespace JT808.DotNetty.Abstractions
+{
+    public interface IJT808MsgReplyConsumer : IJT808PubSub, IDisposable
+    {
+        void OnMessage(Action<(string TerminalNo, byte[] Data)> callback);
+        CancellationTokenSource Cts { get; }
+        void Subscribe();
+        void Unsubscribe();
+    }
+}

src/JT808.DotNetty.Abstractions/IJT808MsgReplyProducer.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JT808.DotNetty.Abstractions
+{
+    public interface IJT808MsgReplyProducer : IJT808PubSub, IDisposable
+    {
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="terminalNo">设备终端号</param>
+        /// <param name="data">808 hex data</param>
+        Task ProduceAsync(string terminalNo, byte[] data);
+    }
+}

src/JT808.DotNetty.Abstractions/IJT808PubSub.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Abstractions
+{
+    public interface IJT808PubSub
+    {
+        string TopicName { get; }
+    }
+}

src/JT808.DotNetty.Abstractions/IJT808SourcePackageDispatcher.cs

@@ -1,15 +0,0 @@
-using System.Threading.Tasks;
-
-namespace JT808.DotNetty.Abstractions
-{
-    /// <summary>
-    /// 源包分发器
-    /// 自定义源包分发器业务
-    /// ConfigureServices:
-    /// services.Replace(new ServiceDescriptor(typeof(IJT808SourcePackageDispatcher),typeof(JT808SourcePackageDispatcherDefaultImpl),ServiceLifetime.Singleton));
-    /// </summary>
-    public interface IJT808SourcePackageDispatcher
-    {
-        Task SendAsync(byte[] data);
-    }
-}

src/JT808.DotNetty.Abstractions/IJT808UplinkPacket.cs

@@ -1,19 +0,0 @@
-using JT808.DotNetty.Abstractions.Enums;
-using System.Threading.Tasks;
-
-namespace JT808.DotNetty.Abstractions
-{
-    /// <summary>
-    /// 上行数据包处理接口
-    /// </summary>
-    public interface IJT808UplinkPacket
-    {
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="data">数据包</param>
-        /// <param name="transportProtocolType">传输协议类型</param>
-        /// <returns></returns>
-        Task ProcessorAsync(byte[] data, JT808TransportProtocolType  transportProtocolType);
-    }
-}

src/JT808.DotNetty.Abstractions/JT808Constants.cs

@@ -6,6 +6,9 @@
 
         public const string SessionOffline = "JT808SessionOffline";
 
+        public const string MsgTopic = "jt808msgdefault";
+        public const string MsgReplyTopic = "jt808msgreplydefault";
+
         public static class JT808WebApiRouteTable
         {
             public const string RouteTablePrefix = "/jt808api";

src/JT808.DotNetty.CleintBenchmark/Services/CleintBenchmarkHostedService.cs

@@ -49,7 +49,6 @@ namespace JT808.DotNetty.CleintBenchmark.Services
             {
                 taskFactory.StartNew((item) => 
                 {
-
                     var client = jT808TcpClientFactory.Create(new DeviceConfig(((int)item).ToString(), clientBenchmarkOptions.IP, clientBenchmarkOptions.Port));
                     int lat = new Random(1000).Next(100000, 180000);
                     int Lng = new Random(1000).Next(100000, 180000);

src/JT808.DotNetty.CleintBenchmark/appsettings.json

@@ -13,8 +13,8 @@
     }
   },
   "ClientBenchmarkOptions": {
-    "IP": "127.0.0.1",
+    "IP": "",
     "Port": 808,
-    "DeviceCount": 10
+    "DeviceCount": 1
   }
 }

src/JT808.DotNetty.Client/JT808.DotNetty.Client.csproj

@@ -21,7 +21,7 @@
     <PackageReference Include="DotNetty.Handlers" Version="0.6.0" />
     <PackageReference Include="DotNetty.Transport.Libuv" Version="0.6.0" />
     <PackageReference Include="DotNetty.Codecs" Version="0.6.0" />
-    <PackageReference Include="JT808" Version="2.1.2" />
+    <PackageReference Include="JT808" Version="2.1.3" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.2.0" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.2.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.2" />

src/JT808.DotNetty.Client/JT808ClientDotnettyExtensions.cs

@@ -3,6 +3,7 @@ using Microsoft.Extensions.DependencyInjection;
 using System;
 using System.Collections.Generic;
 using System.Text;
+using JT808.Protocol;
 
 namespace JT808.DotNetty.Client
 {
@@ -10,6 +11,7 @@ namespace JT808.DotNetty.Client
     {
         public static IServiceCollection AddJT808Client(this IServiceCollection serviceDescriptors)
         {
+            serviceDescriptors.AddJT808Configure();
             serviceDescriptors.AddSingleton<JT808SendAtomicCounterService>();
             serviceDescriptors.AddSingleton<JT808ReceiveAtomicCounterService>();
             serviceDescriptors.AddSingleton<IJT808TcpClientFactory, JT808TcpClientFactory>();

src/JT808.DotNetty.Client/JT808TcpClient.cs

@@ -4,20 +4,16 @@ using DotNetty.Handlers.Timeout;
 using DotNetty.Transport.Bootstrapping;
 using DotNetty.Transport.Channels;
 using DotNetty.Transport.Channels.Sockets;
-using DotNetty.Transport.Libuv;
 using JT808.DotNetty.Client.Handlers;
 using Microsoft.Extensions.Logging;
 using System;
-using System.Collections.Generic;
 using System.Runtime.InteropServices;
-using System.Text;
 using Microsoft.Extensions.DependencyInjection;
 using System.Net;
-using System.Threading.Tasks;
 using JT808.DotNetty.Client.Metadata;
 using JT808.DotNetty.Client.Codecs;
 using JT808.DotNetty.Client.Services;
-using JT808.Protocol.Interfaces;
+using JT808.Protocol;
 
 namespace JT808.DotNetty.Client
 {

src/JT808.DotNetty.Core/Handlers/JT808MsgIdTcpHandlerBase.cs

@@ -1,154 +0,0 @@
-using System;
-using System.Collections.Generic;
-using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Metadata;
-using JT808.DotNetty.Core.Session;
-using JT808.Protocol.Enums;
-using JT808.Protocol.Extensions;
-using JT808.Protocol.MessageBody;
-
-namespace JT808.DotNetty.Core.Handlers
-{
-    /// <summary>
-    /// 基于Tcp模式抽象消息处理业务
-    /// 自定义消息处理业务
-    /// 注意:
-    /// 1.ConfigureServices:
-    /// services.Replace(new ServiceDescriptor(typeof(JT808MsgIdTcpHandlerBase),typeof(JT808MsgIdCustomTcpHandlerImpl),ServiceLifetime.Singleton));
-    /// 2.解析具体的消息体，具体消息调用具体的JT808Serializer.Deserialize<T>
-    /// </summary>
-    public abstract class JT808MsgIdTcpHandlerBase
-    {
-        protected JT808SessionManager sessionManager { get; }
-        /// <summary>
-        /// 初始化消息处理业务
-        /// </summary>
-        protected JT808MsgIdTcpHandlerBase(JT808SessionManager sessionManager)
-        {
-            this.sessionManager = sessionManager;
-            HandlerDict = new Dictionary<ushort, Func<JT808Request, IJT808Reply>>
-            {
-                {JT808MsgId.终端通用应答.ToUInt16Value(), Msg0x0001},
-                {JT808MsgId.终端鉴权.ToUInt16Value(), Msg0x0102},
-                {JT808MsgId.终端心跳.ToUInt16Value(), Msg0x0002},
-                {JT808MsgId.终端注销.ToUInt16Value(), Msg0x0003},
-                {JT808MsgId.终端注册.ToUInt16Value(), Msg0x0100},
-                {JT808MsgId.位置信息汇报.ToUInt16Value(),Msg0x0200 },
-                {JT808MsgId.定位数据批量上传.ToUInt16Value(),Msg0x0704 },
-                {JT808MsgId.数据上行透传.ToUInt16Value(),Msg0x0900 }
-            };
-        }
-
-        public Dictionary<ushort, Func<JT808Request, IJT808Reply>> HandlerDict { get; protected set; }
-        /// <summary>
-        /// 终端通用应答
-        /// 平台无需回复
-        /// 实现自己的业务
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0001(JT808Request request)
-        {
-            return null;
-        }
-        /// <summary>
-        /// 终端心跳
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0002(JT808Request request)
-        {
-            sessionManager.Heartbeat(request.Package.Header.TerminalPhoneNo);
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端注销
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0003(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端注册
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0100(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.终端注册应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8100()
-            {
-                Code = "J" + request.Package.Header.TerminalPhoneNo,
-                JT808TerminalRegisterResult = JT808TerminalRegisterResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端鉴权
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0102(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 位置信息汇报
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0200(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 定位数据批量上传
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0704(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId =request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 数据上行透传
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0900(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId =request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-    }
-}

src/JT808.DotNetty.Core/Handlers/JT808MsgIdUdpHandlerBase.cs

@@ -1,155 +0,0 @@
-using System;
-using System.Collections.Generic;
-using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Metadata;
-using JT808.DotNetty.Core.Session;
-using JT808.Protocol.Enums;
-using JT808.Protocol.Extensions;
-using JT808.Protocol.MessageBody;
-
-namespace JT808.DotNetty.Core.Handlers
-{
-    /// <summary>
-    /// 基于Udp模式的抽象消息处理业务
-    /// 自定义消息处理业务
-    /// 注意:
-    /// 1.ConfigureServices:
-    /// services.Replace(new ServiceDescriptor(typeof(JT808MsgIdUdpHandlerBase),typeof(JT808MsgIdCustomUdpHandlerImpl),ServiceLifetime.Singleton));
-    /// 2.解析具体的消息体，具体消息调用具体的JT808Serializer.Deserialize<T>
-    /// </summary>
-    public abstract class JT808MsgIdUdpHandlerBase
-    {
-        protected JT808SessionManager sessionManager { get; }
-        /// <summary>
-        /// 初始化消息处理业务
-        /// </summary>
-        protected JT808MsgIdUdpHandlerBase(JT808SessionManager sessionManager)
-        {
-            this.sessionManager = sessionManager;
-            HandlerDict = new Dictionary<ushort, Func<JT808Request, IJT808Reply>>
-            {
-                {JT808MsgId.终端通用应答.ToUInt16Value(), Msg0x0001},
-                {JT808MsgId.终端鉴权.ToUInt16Value(), Msg0x0102},
-                {JT808MsgId.终端心跳.ToUInt16Value(), Msg0x0002},
-                {JT808MsgId.终端注销.ToUInt16Value(), Msg0x0003},
-                {JT808MsgId.终端注册.ToUInt16Value(), Msg0x0100},
-                {JT808MsgId.位置信息汇报.ToUInt16Value(),Msg0x0200 },
-                {JT808MsgId.定位数据批量上传.ToUInt16Value(),Msg0x0704 },
-                {JT808MsgId.数据上行透传.ToUInt16Value(),Msg0x0900 }
-            };
-        }
-
-        public Dictionary<ushort, Func<JT808Request, IJT808Reply>> HandlerDict { get; protected set; }
-        /// <summary>
-        /// 终端通用应答
-        /// 平台无需回复
-        /// 实现自己的业务
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0001(JT808Request request)
-        {
-            return null;
-        }
-        /// <summary>
-        /// 终端心跳
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0002(JT808Request request)
-        {
-            sessionManager.Heartbeat(request.Package.Header.TerminalPhoneNo);
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端注销
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0003(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端注册
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0100(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.终端注册应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8100()
-            {
-                Code = "J" + request.Package.Header.TerminalPhoneNo,
-                JT808TerminalRegisterResult = JT808TerminalRegisterResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 终端鉴权
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0102(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 位置信息汇报
-        /// </summary>
-        /// <param name="reqJT808Package"></param>
-        /// <param name="ctx"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0200(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId = request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 定位数据批量上传
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0704(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId =request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-        /// <summary>
-        /// 数据上行透传
-        /// </summary>
-        /// <param name="request"></param>
-        /// <returns></returns>
-        public virtual IJT808Reply Msg0x0900(JT808Request request)
-        {
-            return new JT808Response(JT808MsgId.平台通用应答.Create(request.Package.Header.TerminalPhoneNo, new JT808_0x8001()
-            {
-                MsgId =request.Package.Header.MsgId,
-                JT808PlatformResult = JT808PlatformResult.成功,
-                MsgNum = request.Package.Header.MsgNum
-            }));
-        }
-    }
-}

src/JT808.DotNetty.Core/Impls/JT808DatagramPacketImpl.cs

@@ -13,16 +13,8 @@ namespace JT808.DotNetty.Core.Impls
 {
     class JT808DatagramPacketImpl : IJT808DatagramPacket
     {
-        private readonly IJT808DownlinkPacket jT808DownlinkPacket;
-        public JT808DatagramPacketImpl(
-            IJT808DownlinkPacket jT808DownlinkPacket)
-        {
-            this.jT808DownlinkPacket = jT808DownlinkPacket;
-        }
-
         public DatagramPacket Create(byte[] message, EndPoint recipient)
         {
-            jT808DownlinkPacket.ProcessorAsync(message, JT808TransportProtocolType.udp);
             return new DatagramPacket(Unpooled.WrappedBuffer(message), recipient);
         }
     }

src/JT808.DotNetty.Core/Impls/JT808MsgProducerDefaultImpl.cs

@@ -0,0 +1,29 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Core.Services;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JT808.DotNetty.Core.Impls
+{
+    internal class JT808MsgProducerDefaultImpl : IJT808MsgProducer
+    {
+        private readonly JT808MsgService JT808MsgService;
+        public string TopicName => JT808Constants.MsgTopic;
+        public JT808MsgProducerDefaultImpl(JT808MsgService jT808MsgService)
+        {
+            JT808MsgService = jT808MsgService;
+        }
+        public void Dispose()
+        {
+            
+        }
+
+        public Task ProduceAsync(string terminalNo, byte[] data)
+        {
+            JT808MsgService.MsgQueue.Add((terminalNo, data));
+            return Task.CompletedTask;
+        }
+    }
+}

src/JT808.DotNetty.Core/Impls/JT808MsgReplyConsumerDefaultImpl.cs

@@ -0,0 +1,193 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Core.Services;
+using JT808.Protocol;
+using JT808.Protocol.Enums;
+using JT808.Protocol.Extensions;
+using JT808.Protocol.MessageBody;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using JT808Constants = JT808.DotNetty.Abstractions.JT808Constants;
+
+namespace JT808.DotNetty.Core.Impls
+{
+    internal class JT808MsgReplyConsumerDefaultImpl : IJT808MsgReplyConsumer
+    {
+        private readonly JT808MsgService JT808MsgService;
+        private readonly JT808Serializer JT808Serializer;
+        private Dictionary<ushort, Func<JT808HeaderPackage, byte[]>> HandlerDict;
+        public JT808MsgReplyConsumerDefaultImpl(
+                IJT808Config jT808Config,
+                JT808MsgService jT808MsgService)
+        {
+            JT808MsgService = jT808MsgService;
+            this.JT808Serializer = jT808Config.GetSerializer();
+            HandlerDict = new Dictionary<ushort, Func<JT808HeaderPackage, byte[]>> {
+                {JT808MsgId.终端通用应答.ToUInt16Value(), Msg0x0001},
+                {JT808MsgId.终端鉴权.ToUInt16Value(), Msg0x0102},
+                {JT808MsgId.终端心跳.ToUInt16Value(), Msg0x0002},
+                {JT808MsgId.终端注销.ToUInt16Value(), Msg0x0003},
+                {JT808MsgId.终端注册.ToUInt16Value(), Msg0x0100},
+                {JT808MsgId.位置信息汇报.ToUInt16Value(),Msg0x0200 },
+                {JT808MsgId.定位数据批量上传.ToUInt16Value(),Msg0x0704 },
+                {JT808MsgId.数据上行透传.ToUInt16Value(),Msg0x0900 }
+            };
+        }
+        public CancellationTokenSource Cts =>new CancellationTokenSource();
+
+        public string TopicName => JT808Constants.MsgReplyTopic;
+
+        public void Dispose()
+        {
+            Cts.Dispose();
+        }
+
+        public void OnMessage(Action<(string TerminalNo, byte[] Data)> callback)
+        {
+            Task.Run(() =>
+            {
+                foreach(var item in JT808MsgService.MsgQueue.GetConsumingEnumerable(Cts.Token))
+                {
+                    try
+                    {
+                        var package = JT808Serializer.HeaderDeserialize(item.Data);
+                        if (HandlerDict.TryGetValue(package.Header.MsgId, out var func))
+                        {
+                            var buffer = func(package);
+                            if (buffer != null)
+                            {
+                                callback((item.TerminalNo, item.Data));
+                            }
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+
+                    }
+                }
+            }, Cts.Token);
+        }
+
+        public void Subscribe()
+        {
+            
+        }
+
+        public void Unsubscribe()
+        {
+            Cts.Cancel();
+        }
+
+        /// <summary>
+        /// 终端通用应答
+        /// 平台无需回复
+        /// 实现自己的业务
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0001(JT808HeaderPackage request)
+        {
+            return null;
+        }
+        /// <summary>
+        /// 终端心跳
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0002(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 终端注销
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0003(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 终端注册
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0100(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.终端注册应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8100()
+            {
+                Code = "J" + request.Header.TerminalPhoneNo,
+                JT808TerminalRegisterResult = JT808TerminalRegisterResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 终端鉴权
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0102(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 位置信息汇报
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0200(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 定位数据批量上传
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0704(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+        /// <summary>
+        /// 数据上行透传
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns></returns>
+        public byte[] Msg0x0900(JT808HeaderPackage request)
+        {
+            return JT808Serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
+            {
+                MsgId = request.Header.MsgId,
+                JT808PlatformResult = JT808PlatformResult.成功,
+                MsgNum = request.Header.MsgNum
+            }));
+        }
+    }
+}

src/JT808.DotNetty.Core/Impls/JT808NettyBuilderDefault.cs

@@ -18,38 +18,26 @@ namespace JT808.DotNetty.Core.Impls
             JT808Builder = builder;
         }
 
-        public IJT808NettyBuilder Replace<T>() where T : IJT808SourcePackageDispatcher
-        {
-            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808SourcePackageDispatcher), typeof(T), ServiceLifetime.Singleton));
-            return this;
-        }
-
         public IJT808Builder Builder()
         {
             return JT808Builder;
         }
 
-        public IJT808NettyBuilder ReplaceSourcePackageDispatcher<T>() where T : IJT808SourcePackageDispatcher
+        public IJT808NettyBuilder ReplaceSessionPublishing<T>() where T : IJT808SessionPublishing
-        {
-            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808SourcePackageDispatcher), typeof(T), ServiceLifetime.Singleton));
-            return this;
-        }
-
-        public IJT808NettyBuilder ReplaceDownlinkPacket<T>() where T : IJT808DownlinkPacket
         {
-            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808DownlinkPacket), typeof(T), ServiceLifetime.Singleton));
+            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808SessionPublishing), typeof(T), ServiceLifetime.Singleton));
             return this;
         }
 
-        public IJT808NettyBuilder ReplaceUplinkPacket<T>() where T : IJT808UplinkPacket
+        public IJT808NettyBuilder ReplaceMsgProducer<T>() where T : IJT808MsgProducer
         {
-            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808UplinkPacket), typeof(T), ServiceLifetime.Singleton));
+            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808MsgProducer), typeof(T), ServiceLifetime.Singleton));
             return this;
         }
 
-        public IJT808NettyBuilder ReplaceSessionPublishing<T>() where T : IJT808SessionPublishing
+        public IJT808NettyBuilder ReplaceMsgReplyConsumer<T>() where T : IJT808MsgReplyConsumer
         {
-            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808SessionPublishing), typeof(T), ServiceLifetime.Singleton));
+            JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808MsgReplyConsumer), typeof(T), ServiceLifetime.Singleton));
             return this;
         }
     }

src/JT808.DotNetty.Core/Impls/JT808SessionPublishingEmptyImpl.cs

@@ -1,12 +1,22 @@
 using JT808.DotNetty.Abstractions;
+using Microsoft.Extensions.Logging;
 using System.Threading.Tasks;
 
 namespace JT808.DotNetty.Core
 {
     internal class JT808SessionPublishingEmptyImpl : IJT808SessionPublishing
     {
+        private readonly ILogger<JT808SessionPublishingEmptyImpl> logger;
+        public JT808SessionPublishingEmptyImpl(ILoggerFactory loggerFactory)
+        {
+            logger = loggerFactory.CreateLogger<JT808SessionPublishingEmptyImpl>();
+        }
         public Task PublishAsync(string topicName, string value)
         {
+            if (logger.IsEnabled(LogLevel.Debug))
+            {
+                logger.LogDebug($"{topicName}-{value}");
+            }
             return Task.CompletedTask;
         }
     }

src/JT808.DotNetty.Core/Interfaces/IJT808NettyBuilder.cs

@@ -11,6 +11,7 @@ namespace JT808.DotNetty.Core.Interfaces
     {
         IJT808Builder JT808Builder { get; }
         IJT808NettyBuilder ReplaceSessionPublishing<T>() where T : IJT808SessionPublishing;
+        IJT808NettyBuilder ReplaceMsgProducer<T>() where T : IJT808MsgProducer;
         IJT808Builder Builder();
     }
 }

src/JT808.DotNetty.Core/Interfaces/IJT808TcpNettyBuilder.cs

@@ -1,13 +0,0 @@
-using JT808.DotNetty.Core.Handlers;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.Core.Interfaces
-{
-    public interface IJT808TcpNettyBuilder
-    {
-        IJT808NettyBuilder Instance { get; }
-        IJT808NettyBuilder Builder();
-    }
-}

src/JT808.DotNetty.Core/Interfaces/IJT808UdpNettyBuilder.cs

@@ -1,13 +0,0 @@
-using JT808.DotNetty.Core.Handlers;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.Core.Interfaces
-{
-    public interface IJT808UdpNettyBuilder
-    {
-        IJT808NettyBuilder Instance { get; }
-        IJT808NettyBuilder Builder();
-    }
-}

src/JT808.DotNetty.Core/JT808.DotNetty.Core.csproj

@@ -11,7 +11,7 @@
     <PackageReference Include="DotNetty.Handlers" Version="0.6.0" />
     <PackageReference Include="DotNetty.Transport.Libuv" Version="0.6.0" />
     <PackageReference Include="DotNetty.Codecs" Version="0.6.0" />
-    <PackageReference Include="JT808" Version="2.1.2" />
+    <PackageReference Include="JT808" Version="2.1.3" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.2.0" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="2.2.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.2" />

src/JT808.DotNetty.Core/JT808CoreDotnettyExtensions.cs

@@ -63,6 +63,11 @@ namespace JT808.DotNetty.Core
             nettyBuilder.JT808Builder.Services.TryAddSingleton<JT808SessionManager>();
             nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808UnificationSendService, JT808UnificationSendService>();
             nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808SessionService, JT808SessionService>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgProducer, JT808MsgProducerDefaultImpl>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgReplyConsumer, JT808MsgReplyConsumerDefaultImpl>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808SessionPublishing, JT808SessionPublishingEmptyImpl>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<JT808MsgService>();
+            nettyBuilder.JT808Builder.Services.AddHostedService<JT808MsgReplyHostedService>();
             return nettyBuilder;
         }
 
@@ -85,6 +90,11 @@ namespace JT808.DotNetty.Core
             nettyBuilder.JT808Builder.Services.TryAddSingleton<JT808SessionManager>();
             nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808UnificationSendService, JT808UnificationSendService>();
             nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808SessionService, JT808SessionService>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgProducer, JT808MsgProducerDefaultImpl>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgReplyConsumer, JT808MsgReplyConsumerDefaultImpl>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<JT808MsgService>();
+            nettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808SessionPublishing, JT808SessionPublishingEmptyImpl>();
+            nettyBuilder.JT808Builder.Services.AddHostedService<JT808MsgReplyHostedService>();
             return nettyBuilder;
         }
     }

src/JT808.DotNetty.Core/Services/JT808MsgReplyHostedService.cs

@@ -0,0 +1,42 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Core.Session;
+using Microsoft.Extensions.Hosting;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JT808.DotNetty.Core.Services
+{
+    internal class JT808MsgReplyHostedService : IHostedService
+    {
+        private readonly JT808SessionManager JT808SessionManager;
+
+        private readonly IJT808MsgReplyConsumer JT808MsgReplyConsumer;
+
+        public JT808MsgReplyHostedService(
+            IJT808MsgReplyConsumer jT808MsgReplyConsumer,
+            JT808SessionManager jT808SessionManager)
+        {
+            JT808MsgReplyConsumer = jT808MsgReplyConsumer;
+            JT808SessionManager = jT808SessionManager;
+        }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            JT808MsgReplyConsumer.OnMessage(item => 
+            {
+                JT808SessionManager.Send(item.TerminalNo, item.Data);
+            });
+            JT808MsgReplyConsumer.Subscribe();
+            return Task.CompletedTask;    
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            JT808MsgReplyConsumer.Unsubscribe();
+            return Task.CompletedTask;
+        }
+    }
+}

src/JT808.DotNetty.Core/Services/JT808MsgService.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JT808.DotNetty.Core.Services
+{
+    internal class JT808MsgService
+    {
+        public System.Collections.Concurrent.BlockingCollection<(string TerminalNo, byte[] Data)> MsgQueue { get; set; } = new System.Collections.Concurrent.BlockingCollection<(string TerminalNo, byte[] Data)>();
+    }
+}

src/JT808.DotNetty.Core/Session/JT808SessionManager.cs

@@ -120,6 +120,21 @@ namespace JT808.DotNetty.Core.Session
             }
             return isSuccessed;
         }
+        public void Send(string terminalPhoneNo, byte[] data)
+        {
+            var session = GetSessionByTerminalPhoneNo(terminalPhoneNo);
+            if (session != null)
+            {
+                if (session.TransportProtocolType == JT808TransportProtocolType.tcp)
+                {
+                    session.Channel.WriteAndFlushAsync(new JT808Response(data));
+                }
+                else if (session.TransportProtocolType == JT808TransportProtocolType.udp)
+                {
+                    session.Channel.WriteAndFlushAsync(jT808DatagramPacket.Create(data, ((JT808UdpSession)session).Sender));
+                } 
+            }
+        }
         public bool TrySend(string terminalPhoneNo, IJT808Reply reply, out string message)
         {
             bool isSuccessed;

src/JT808.DotNetty.Tcp/Handlers/JT808TcpServerHandler.cs

@@ -9,6 +9,7 @@ using JT808.DotNetty.Abstractions.Enums;
 using JT808.Protocol.Interfaces;
 using JT808.Protocol.Exceptions;
 using JT808.DotNetty.Core.Session;
+using JT808.DotNetty.Abstractions;
 
 namespace JT808.DotNetty.Tcp.Handlers
 {
@@ -25,13 +26,17 @@ namespace JT808.DotNetty.Tcp.Handlers
 
         private readonly JT808Serializer JT808Serializer;
 
+        private readonly IJT808MsgProducer JT808MsgProducer;
+
         public JT808TcpServerHandler(
+            IJT808MsgProducer jT808MsgProducer,
             IJT808Config jT808Config,
             ILoggerFactory loggerFactory,
             JT808AtomicCounterServiceFactory  jT808AtomicCounterServiceFactory,
             JT808SessionManager jT808SessionManager)
         {
             this.jT808SessionManager = jT808SessionManager;
+            this.JT808MsgProducer = jT808MsgProducer;
             this.jT808AtomicCounterService = jT808AtomicCounterServiceFactory.Create(JT808TransportProtocolType.tcp);
             this.JT808Serializer = jT808Config.GetSerializer();
             logger = loggerFactory.CreateLogger<JT808TcpServerHandler>();
@@ -51,6 +56,7 @@ namespace JT808.DotNetty.Tcp.Handlers
                     logger.LogTrace($"accept package success count=>{jT808AtomicCounterService.MsgSuccessCount.ToString()},accept msg=>{ByteBufferUtil.HexDump(msg)}");
                 }
                 jT808SessionManager.TryAdd(jT808HeaderPackage.Header.TerminalPhoneNo,ctx.Channel);
+                JT808MsgProducer.ProduceAsync(jT808HeaderPackage.Header.TerminalPhoneNo, msg);
             }
             catch (JT808Exception ex)
             {

src/JT808.DotNetty.Tcp/JT808TcpBuilderDefault.cs

@@ -1,25 +0,0 @@
-using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Core.Interfaces;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.Tcp
-{
-    public class JT808TcpBuilderDefault : IJT808TcpNettyBuilder
-    {
-        public IJT808NettyBuilder Instance { get; }
-
-        public JT808TcpBuilderDefault(IJT808NettyBuilder builder)
-        {
-            Instance = builder;
-        }
-
-        public IJT808NettyBuilder Builder()
-        {
-            return Instance;
-        }
-    }
-}

src/JT808.DotNetty.Tcp/JT808TcpDotnettyExtensions.cs

@@ -1,17 +1,9 @@
 using JT808.DotNetty.Core.Codecs;
-using JT808.DotNetty.Core;
-using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Core.Services;
 using JT808.DotNetty.Tcp.Handlers;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
-using Microsoft.Extensions.Hosting;
-using Newtonsoft.Json;
-using System;
-using System.Reflection;
 using System.Runtime.CompilerServices;
 using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Internal;
 
 [assembly: InternalsVisibleTo("JT808.DotNetty.Tcp.Test")]
 
@@ -19,14 +11,14 @@ namespace JT808.DotNetty.Tcp
 {
     public static class JT808TcpDotnettyExtensions
     {
-        public static IJT808TcpNettyBuilder AddJT808TcpNettyHost(this IJT808NettyBuilder jT808NettyBuilder)
+        public static IJT808NettyBuilder AddJT808TcpNettyHost(this IJT808NettyBuilder jT808NettyBuilder)
         {
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808TcpConnectionHandler>();
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808TcpEncoder>();
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808TcpDecoder>();
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808TcpServerHandler>();
             jT808NettyBuilder.JT808Builder.Services.AddHostedService<JT808TcpServerHost>();
-            return new JT808TcpBuilderDefault(jT808NettyBuilder);
+            return jT808NettyBuilder;
         }
     }
 }

src/JT808.DotNetty.Tests/JT808.DotNetty.Core.Test/SeedTcpSession.cs

@@ -13,7 +13,7 @@ namespace JT808.DotNetty.Core.Test
     public class SeedTcpSession
     {
         public JT808SessionManager jT80TcpSessionManager = new JT808SessionManager(
-            new JT808SessionPublishingEmptyImpl(),
+            new JT808SessionPublishingEmptyImpl(new LoggerFactory()),
             new LoggerFactory());
 
         public SeedTcpSession()

src/JT808.DotNetty.Tests/JT808.DotNetty.Hosting/Handlers/JT808MsgIdTcpCustomHandler.cs

@@ -1,60 +0,0 @@
-using JT808.DotNetty.Core;
-using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Metadata;
-using JT808.DotNetty.Core.Session;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.Hosting.Handlers
-{
-    public class JT808MsgIdTcpCustomHandler : JT808MsgIdTcpHandlerBase
-    {
-        public JT808MsgIdTcpCustomHandler(
-            ILoggerFactory loggerFactory,
-            JT808SessionManager sessionManager) : base(sessionManager)
-        {
-            logger = loggerFactory.CreateLogger<JT808MsgIdTcpCustomHandler>();
-        }
-
-        private readonly ILogger<JT808MsgIdTcpCustomHandler> logger;
-
-        public override IJT808Reply Msg0x0200(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0200");
-            return base.Msg0x0200(request);
-        }
-
-        public override IJT808Reply Msg0x0001(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0001");
-            return base.Msg0x0001(request);
-        }
-
-        public override IJT808Reply Msg0x0002(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0002");
-            return base.Msg0x0002(request);
-        }
-
-        public override IJT808Reply Msg0x0003(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0003");
-            return base.Msg0x0003(request);
-        }
-
-        public override IJT808Reply Msg0x0100(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0100");
-            return base.Msg0x0100(request);
-        }
-
-        public override IJT808Reply Msg0x0102(JT808Request request)
-        {
-            logger.LogDebug("Tcp_Msg0x0102");
-            return base.Msg0x0102(request);
-        }
-    }
-}

src/JT808.DotNetty.Tests/JT808.DotNetty.Hosting/Handlers/JT808MsgIdUdpCustomHandler.cs

@@ -1,60 +0,0 @@
-using JT808.DotNetty.Core;
-using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Metadata;
-using JT808.DotNetty.Core.Session;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.DotNetty.Hosting.Handlers
-{
-    public class JT808MsgIdUdpCustomHandler : JT808MsgIdUdpHandlerBase
-    {
-        public JT808MsgIdUdpCustomHandler(
-            ILoggerFactory loggerFactory,
-            JT808SessionManager sessionManager) : base(sessionManager)
-        {
-            logger = loggerFactory.CreateLogger<JT808MsgIdUdpCustomHandler>();
-        }
-        
-        private readonly ILogger<JT808MsgIdUdpCustomHandler> logger;
-
-        public override IJT808Reply Msg0x0200(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0200");
-            return base.Msg0x0200(request);
-        }
-
-        public override IJT808Reply Msg0x0001(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0001");
-            return base.Msg0x0001(request);
-        }
-
-        public override IJT808Reply Msg0x0002(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0002");
-            return base.Msg0x0002(request);
-        }
-
-        public override IJT808Reply Msg0x0003(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0003");
-            return base.Msg0x0003(request);
-        }
-
-        public override IJT808Reply Msg0x0100(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0100");
-            return base.Msg0x0100(request);
-        }
-
-        public override IJT808Reply Msg0x0102(JT808Request request)
-        {
-            logger.LogDebug("Udp_Msg0x0102");
-            return base.Msg0x0102(request);
-        }
-    }
-}

src/JT808.DotNetty.Tests/JT808.DotNetty.Hosting/Impls/JT808DownlinkPacketLogging.cs

@@ -1,25 +0,0 @@
-using JT808.DotNetty.Abstractions;
-using JT808.DotNetty.Abstractions.Enums;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
-using JT808.Protocol.Extensions;
-
-namespace JT808.DotNetty.Hosting.Impls
-{
-    public class JT808DownlinkPacketLogging : IJT808DownlinkPacket
-    {
-        private readonly ILogger logger;
-        public JT808DownlinkPacketLogging(ILoggerFactory loggerFactory )
-        {
-            logger = loggerFactory.CreateLogger("JT808DownlinkPacketLogging");
-        }
-        public Task ProcessorAsync(byte[] data, JT808TransportProtocolType transportProtocolType)
-        {
-           logger.LogInformation("send >>>"+data.ToHexString());
-           return Task.CompletedTask;
-        }
-    }
-}

src/JT808.DotNetty.Tests/JT808.DotNetty.Hosting/Program.cs

@@ -1,8 +1,6 @@
 using JT808.DotNetty.Abstractions;
 using JT808.DotNetty.Core;
 using JT808.DotNetty.Core.Handlers;
-using JT808.DotNetty.Hosting.Handlers;
-using JT808.DotNetty.Hosting.Impls;
 using JT808.DotNetty.Tcp;
 using JT808.DotNetty.Udp;
 using JT808.DotNetty.WebApi;
@@ -55,20 +53,18 @@ namespace JT808.DotNetty.Hosting
                     services.AddJT808Configure()
                             .AddJT808NettyCore(hostContext.Configuration)
                             .AddJT808TcpNettyHost()
-                            .Builder()
+                            //.AddJT808UdpNettyHost()
-                            .AddJT808UdpNettyHost()
-                            .Builder()
                             .AddJT808WebApiNettyHost()
                             .Builder();
                     //webapi客户端调用
-                    services.AddHttpApi<IJT808DotNettyWebApi>().ConfigureHttpApiConfig((c, p) =>
+                    //services.AddHttpApi<IJT808DotNettyWebApi>().ConfigureHttpApiConfig((c, p) =>
-                    {
+                    //{
-                        c.HttpHost = new Uri("http://localhost:828/jt808api/");
+                    //    c.HttpHost = new Uri("http://localhost:828/jt808api/");
-                        c.FormatOptions.DateTimeFormat = "yyyy-MM-dd HH:mm:ss.fff";
+                    //    c.FormatOptions.DateTimeFormat = "yyyy-MM-dd HH:mm:ss.fff";
-                        c.LoggerFactory = p.GetRequiredService<ILoggerFactory>();
+                    //    c.LoggerFactory = p.GetRequiredService<ILoggerFactory>();
-                    });
+                    //});
-                    var client = services.BuildServiceProvider().GetRequiredService<IJT808DotNettyWebApi>();
+                    //var client = services.BuildServiceProvider().GetRequiredService<IJT808DotNettyWebApi>();
-                    var result = client.GetTcpAtomicCounter().InvokeAsync().Result;
+                    //var result = client.GetTcpAtomicCounter().InvokeAsync().Result;
                 });
 
             await serverHostBuilder.RunConsoleAsync();

src/JT808.DotNetty.Tests/JT808.DotNetty.Hosting/appsettings.json

@@ -15,9 +15,6 @@
   "JT808Configuration": {
     "TcpPort": 808,
     "UdpPort": 808,
-    "WebApiPort": 828,
+    "WebApiPort": 828
-    "ForwardingRemoteIPAddress": [
-      //"172.16.19.209"
-    ]
   }
 }

src/JT808.DotNetty.Tests/JT808.DotNetty.WebApi.Test/TestBase.cs

@@ -38,9 +38,7 @@ namespace JT808.DotNetty.WebApi.Test
                     services.AddJT808Configure()
                             .AddJT808NettyCore(hostContext.Configuration)
                             .AddJT808TcpNettyHost()
-                            .Builder()
                             .AddJT808UdpNettyHost()
-                            .Builder()
                             .AddJT808WebApiNettyHost();
                 });
             var build = serverHostBuilder.Build();

src/JT808.DotNetty.Udp/Handlers/JT808UdpServerHandler.cs

@@ -9,6 +9,7 @@ using JT808.DotNetty.Core;
 using JT808.DotNetty.Abstractions.Enums;
 using JT808.Protocol.Interfaces;
 using JT808.DotNetty.Core.Session;
+using JT808.DotNetty.Abstractions;
 
 namespace JT808.DotNetty.Udp.Handlers
 {
@@ -25,12 +26,16 @@ namespace JT808.DotNetty.Udp.Handlers
 
         private readonly JT808Serializer JT808Serializer;
 
+        private readonly IJT808MsgProducer JT808MsgProducer;
+
         public JT808UdpServerHandler(
+            IJT808MsgProducer jT808MsgProducer,
             IJT808Config jT808Config,
             ILoggerFactory loggerFactory,
             JT808AtomicCounterServiceFactory  jT808AtomicCounterServiceFactory,
             JT808SessionManager jT808UdpSessionManager)
         {
+            this.JT808MsgProducer = jT808MsgProducer;
             this.jT808AtomicCounterService = jT808AtomicCounterServiceFactory.Create(JT808TransportProtocolType.udp);
             this.jT808UdpSessionManager = jT808UdpSessionManager;
             logger = loggerFactory.CreateLogger<JT808UdpServerHandler>();
@@ -47,6 +52,7 @@ namespace JT808.DotNetty.Udp.Handlers
                 JT808HeaderPackage jT808HeaderPackage = JT808Serializer.Deserialize<JT808HeaderPackage>(msg.Buffer);
                 jT808AtomicCounterService.MsgSuccessIncrement();
                 jT808UdpSessionManager.TryAdd(ctx.Channel, msg.Sender, jT808HeaderPackage.Header.TerminalPhoneNo);
+                JT808MsgProducer.ProduceAsync(jT808HeaderPackage.Header.TerminalPhoneNo, msg.Buffer);
                 if (logger.IsEnabled(LogLevel.Trace))
                 {
                     logger.LogTrace($"accept package success count=>{jT808AtomicCounterService.MsgFailCount.ToString()},accept msg=>{ByteBufferUtil.HexDump(msg.Buffer)}");

src/JT808.DotNetty.Udp/JT1078UdpBuilderDefault.cs

@@ -1,21 +0,0 @@
-using JT808.DotNetty.Core.Interfaces;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-
-namespace JT808.DotNetty.Udp
-{
-    class JT1078UdpBuilderDefault : IJT808UdpNettyBuilder
-    {
-        public IJT808NettyBuilder Instance { get; }
-
-        public JT1078UdpBuilderDefault(IJT808NettyBuilder builder)
-        {
-            Instance = builder;
-        }
-
-        public IJT808NettyBuilder Builder()
-        {
-            return Instance;
-        }
-    }
-}

src/JT808.DotNetty.Udp/JT808UdpDotnettyExtensions.cs

@@ -1,14 +1,9 @@
-using JT808.DotNetty.Core;
+using JT808.DotNetty.Core.Codecs;
-using JT808.DotNetty.Core.Codecs;
-using JT808.DotNetty.Core.Handlers;
 using JT808.DotNetty.Core.Impls;
 using JT808.DotNetty.Core.Interfaces;
-using JT808.DotNetty.Core.Services;
-using JT808.DotNetty.Internal;
 using JT808.DotNetty.Udp.Handlers;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
-using System;
 using System.Runtime.CompilerServices;
 
 [assembly: InternalsVisibleTo("JT808.DotNetty.Udp.Test")]
@@ -17,13 +12,13 @@ namespace JT808.DotNetty.Udp
 {
     public static class JT808UdpDotnettyExtensions
     {
-        public static IJT808UdpNettyBuilder AddJT808UdpNettyHost(this IJT808NettyBuilder jT808NettyBuilder)
+        public static IJT808NettyBuilder AddJT808UdpNettyHost(this IJT808NettyBuilder jT808NettyBuilder)
         {
             jT808NettyBuilder.JT808Builder.Services.TryAddSingleton<IJT808DatagramPacket, JT808DatagramPacketImpl>();
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808UdpDecoder>();
             jT808NettyBuilder.JT808Builder.Services.TryAddScoped<JT808UdpServerHandler>();
             jT808NettyBuilder.JT808Builder.Services.AddHostedService<JT808UdpServerHost>();
-            return new JT1078UdpBuilderDefault(jT808NettyBuilder);
+            return jT808NettyBuilder;
         }
     }
 }

src/SharedProperties.props

@@ -9,7 +9,7 @@
     <licenseUrl>https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE</licenseUrl>
     <license>https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE</license>
     <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>2.1.2</Version>
+    <Version>2.2.0</Version>
     <SignAssembly>false</SignAssembly>
     <PackageLicenseFile>LICENSE</PackageLicenseFile>
     <PackageRequireLicenseAcceptance>true</PackageRequireLicenseAcceptance>