Merge branch 'master' of https://github.com/SmallChi/JT808DotNetty

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2018 SmallChi(Koike)
+Copyright (c) 2019 SmallChi(Koike)
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -1,6 +1,8 @@
-# JT808DotNetty
+# JT808Gateway
 
-基于DotNetty封装的JT808DotNetty支持TCP/UDP通用消息业务处理 
+基于DotNetty封装的JT808DotNetty支持TCP/UDP通用消息业务处理
+
+基于Pipeline封装的JT808Pipeline支持TCP/UDP通用消息业务处理
 
 [了解JT808协议进这边](https://github.com/SmallChi/JT808)
 
@@ -14,7 +16,7 @@
 
 [![MIT Licence](https://img.shields.io/github/license/mashape/apistatus.svg)](https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE)[![Build Status](https://travis-ci.org/SmallChi/JT808DotNetty.svg?branch=master)](https://travis-ci.org/SmallChi/JT808DotNetty)
 
-## 新网关的优势：
+## 新网关的优势
 
 1. 跨平台
 2. 借助 .NET Core模块化的思想
@@ -55,7 +57,7 @@
 |Traffic|流量统计服务 |由于运营商sim卡查询流量滞后，通过流量统计服务可以实时准确的统计设备流量，可以最优配置设备的流量大小，以节省成本
 |Transmit| 原包转发服务|该服务可以将设备上报原始数据转发到第三方，支持全部转发，指定终端号转发|
 
-## NuGet安装
+## 基于DotNetty的NuGet安装
 
 | Package Name          | Version                                            | Downloads                                           |
 | --------------------- | -------------------------------------------------- | --------------------------------------------------- |
@@ -75,6 +77,14 @@
 | Install-Package JT808.DotNetty.Kafka | ![JT808](https://img.shields.io/nuget/v/JT808.DotNetty.Kafka.svg) | ![JT808](https://img.shields.io/nuget/dt/JT808.DotNetty.Kafka.svg) |
 | Install-Package JT808.DotNetty.RabbitMQ | ![JT808](https://img.shields.io/nuget/v/JT808.DotNetty.RabbitMQ.svg) | ![JT808](https://img.shields.io/nuget/dt/JT808.DotNetty.RabbitMQ.svg) |
 
+## 基于Pipeline的NuGet安装
+
+| Package Name          | Version                                            | Downloads                                           |
+| --------------------- | -------------------------------------------------- | --------------------------------------------------- |
+| Install-Package JT808.Gateway.Abstractions| ![JT808.Gateway.Abstractions](https://img.shields.io/nuget/v/JT808.Gateway.Abstractions.svg) | ![JT808.Gateway.Abstractions](https://img.shields.io/nuget/dt/JT808.Gateway.Abstractions.svg) |
+| Install-Package JT808.Gateway | ![JT808.Gateway](https://img.shields.io/nuget/v/JT808.Gateway.svg) | ![JT808.Gateway](https://img.shields.io/nuget/dt/JT808.Gateway.svg) |
+| Install-Package JT808.Gateway.Kafka| ![JT808.Gateway.Kafka](https://img.shields.io/nuget/v/JT808.Gateway.Kafka.svg) | ![JT808.Gateway.Kafka](https://img.shields.io/nuget/dt/JT808.Gateway.Kafka.svg) |
+
 ## 举个栗子1
 
 ``` demo1
@@ -124,7 +134,7 @@ static async Task Main(string[] args)
 ```
 
 如图所示：
-![demo1](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/demo1.png)
+![demo1](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/demo1.png)
 
 ## 举个栗子2
 
@@ -135,4 +145,4 @@ static async Task Main(string[] args)
 3.进入JT808.DotNetty.SimpleClient项目下的Debug目录运行客户端
 
 如图所示：
-![demo2](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/demo2.png)
+![demo2](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/demo2.png)

README_V1.0.0.md

@@ -1,120 +0,0 @@
-# JT808DotNetty
-
-基于DotNetty封装的JT808DotNetty支持TCP/UDP通用消息业务处理
-
-[了解JT808协议进这边](https://github.com/SmallChi/JT808)
-
-[了解JT809协议进这边](https://github.com/SmallChi/JT809)
-
-[了解JT1078协议进这边](https://github.com/SmallChi/JT1078)
-
-[了解JTNE协议进这边](https://github.com/SmallChi/JTNewEnergy)
-
-[玩一玩压力测试](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/README.md)
-
-[V2.2.1老版本](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/README_V2.2.1.md)
-
-[![MIT Licence](https://img.shields.io/github/license/mashape/apistatus.svg)](https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE)
-
-## 新网关的优势
-
-1. 跨平台
-2. 借助 .NET Core模块化的思想
-3. 单机同时一万辆车在线不是梦(真有一万辆车那都很吃香了<(￣3￣)> <(￣3￣)> <(￣3￣)>  )
-4. 简单易上手
-
-## 设计模型
-
-![design_model](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/design_model.png)
-
-## 基于Grpc的消息业务处理程序(JT808.Gateway.GrpcService)
-
-``` 1
-services.AddGrpcClient<JT808Gateway.JT808GatewayClient>(o =>
-{
-    o.Address = new Uri("https://localhost:5001");
-});
-```
-
-## 集成接口功能(JT808.Gateway.PubSub)
-
-|接口名称|接口说明|使用场景|
-|:------:|:------|:------|
-| IJT808SessionProducer| 会话通知（在线/离线）数据生产接口| 有些超长待机的设备，不会实时保持连接，那么通过平台下发的命令是无法到达的，这时候就需要设备一上线，就即时通知服务去处理，然后在即时的下发消息到设备。|
-| IJT808SessionConsumer| 会话通知（在线/离线）数据消费接口| -|
-| IJT808MsgProducer| 数据生产接口| 网关将接收到的数据发送到队列|
-| IJT808MsgConsumer| 数据消费接口| 将数据进行对应的消息业务处理(例：设备流量统计、第三方平台数据转发、消息日志等) |
-| IJT808MsgReplyProducer| 应答数据生产接口|将生产的数据解析为对应的消息Id应答发送到队列 |
-| IJT808MsgReplyConsumer| 应答数据消费接口| 将接收到的应答数据下发给设备|
-
-> 使用物联网卡通过udp下发指令时，存储的那个socket地址端口，有效期非常短,不速度快点下发，那个socket地址端口就可能映射到别的对应卡去了,所以此处采用跟随设备消息下发指令。
-
-## 基于网关的相关服务(JT808.Gateway.BusinessServices)
-
-|服务名称|服务说明|使用场景|
-|:------:|:------|:------|
-|MsgIdHandler| 消息处理服务|从队列中消费设备上报数据，再结合自身的业务场景，将数据进行处理并入库 |
-|MsgLogging | 消息日志服务|从队列中消费设备上报和平台应答数据，再将数据存入influxdb等数据库中，便于技术和技术支持排查设备与平台交互的原始数据|
-|ReplyMessage| 消息响应服务| 用于响应设备上报消息，以及下发指令信息到设备|
-|SessionNotice| 会话管理服务| 通知设备上线下线，对于udp设备来说，可以在设备上线时，将指令跟随消息下发到设备|
-|Traffic|流量统计服务 |由于运营商sim卡查询流量滞后，通过流量统计服务可以实时准确的统计设备流量，可以最优配置设备的流量大小，以节省成本
-|Transmit| 原包转发服务|该服务可以将设备上报原始数据转发到第三方，支持全部转发，指定终端号转发|
-
-## NuGet安装
-
-| Package Name          | Version                                            | Downloads                                           |
-| --------------------- | -------------------------------------------------- | --------------------------------------------------- |
-| Install-Package JT808.Gateway | ![JT808.Gateway](https://img.shields.io/nuget/v/JT808.Gateway.svg) | ![JT808.Gateway](https://img.shields.io/nuget/dt/JT808.Gateway.svg) |
-| Install-Package JT808.Gateway.Kafka| ![JT808.Gateway.Kafka](https://img.shields.io/nuget/v/JT808.Gateway.Kafka.svg) | ![JT808.Gateway.Kafka](https://img.shields.io/nuget/dt/JT808.Gateway.Kafka.svg) |
-
-## 举个栗子1
-
-1.进入JT808.Gateway.SimpleServer项目下的Debug目录运行服务端
-
-2.进入JT808.Gateway.SimpleClient项目下的Debug目录运行客户端
-
-``` 1
-static void Main(string[] args)
-{
-    Host.CreateDefaultBuilder(args)
-        .ConfigureWebHostDefaults(webBuilder =>
-        {
-            //配置Grpc服务端
-            webBuilder
-                .ConfigureKestrel(options =>
-                {
-                    options.Listen(IPAddress.Any, 5001, listenOptions =>
-                    {
-                        listenOptions.Protocols = HttpProtocols.Http2;
-                        listenOptions.UseHttps($"{Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Configs", "test.cer")}", "");
-                    });
-                })
-                .Configure(app =>
-                {
-                    app.UseRouting();
-                    app.UseEndpoints(endpoints =>
-                    {
-                        //配置Grpc服务节点
-                        endpoints.MapGrpcService<JT808GatewayService>();
-                    });
-                });
-        })
-        .ConfigureServices((hostContext,services) =>
-        {
-            //services.Configure<KestrelServerOptions>(hostContext.Configuration.GetSection("Kestrel"));
-            //添加Grpc服务
-            services.AddGrpc();
-            //添加JT808协议
-            services.AddJT808Configure()
-                    //添加JT808网关配置
-                    .AddJT808Gateway(hostContext.Configuration)
-                    //添加基于Tcp的808网关
-                    .AddJT808GatewayTcpHost()
-                    //添加基于Udp的808网关
-                    .AddJT808GatewayUdpHost()
-                    .Builder();
-        })
-        .Build()
-        .Run();
-}
-```

doc/README.md

@@ -1,4 +1,6 @@
-## 压力测试
+# 压力测试
+
+## 基于DotNetty
 
 [感谢泥水佬提供的压力测试工具](https://www.cnblogs.com/smark/p/4496660.html?utm_source=tuicool)
 
@@ -7,10 +9,25 @@
 | win server 2016 | 4c8g | 压力测试客户端 |
 | centos7 | 4c8g | JT808服务端 |
 
-![performance_1000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/performance_1000.png)
+![performance_1000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/performance_1000.png)
+
+![performance_2000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/performance_2000.png)
+
+![performance_5000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/performance_5000.png)
+
+![performance_10000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/dotnetty/performance_10000.png)
+
+## 基于pipeline
+
+| 操作系统 | 配置 | 使用 |
+|:-------:|:-------:|:-------:|
+| centos7 | 4c8g | JT808服务端 |
+| centos7 | 4c8g | JT808客户端 |
+
+> 计算网络增强型 sn1ne ecs.sn1ne.xlarge 4 vCPU 8 GiB Intel Xeon E5-2682v4 / Intel Xeon(Skylake) Platinum 8163 2.5 GHz 1.5 Gbps 50 万 PPS
 
-![performance_2000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/performance_2000.png)
+![server_proccess_10k](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/pipeline/server_proccess_10k.png)
 
-![performance_5000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/performance_5000.png)
+![server_network_10k](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/pipeline/server_network_10k.png)
 
-![performance_10000](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/img/performance_10000.png)
+![client_10k](https://github.com/SmallChi/JT808DotNetty/blob/master/doc/pipeline/client_10k.png)

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

@@ -24,6 +24,7 @@ namespace JT808.DotNetty.SimpleClient.Services
         {
             string sim = "22222222222";
             JT808TcpClient client1 = jT808TcpClientFactory.Create(new JT808DeviceConfig(sim, "127.0.0.1", 808, JT808Version.JTT2019));
+            Thread.Sleep(5000);
             //1.终端注册
             client1.Send(JT808MsgId.终端注册.Create2019(sim, new JT808_0x0100()
             {

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

@@ -24,6 +24,7 @@ namespace JT808.DotNetty.SimpleClient.Services
         {
             string sim = "11111111111";
             JT808TcpClient client1 = jT808TcpClientFactory.Create(new JT808DeviceConfig(sim, "127.0.0.1", 808));
+            Thread.Sleep(5000);
             //1.终端注册
             client1.Send(JT808MsgId.终端注册.Create(sim, new JT808_0x0100()
             {

src/.dockerignore

@@ -0,0 +1,25 @@
+**/.classpath
+**/.dockerignore
+**/.env
+**/.git
+**/.gitignore
+**/.project
+**/.settings
+**/.toolstarget
+**/.vs
+**/.vscode
+**/*.*proj.user
+**/*.dbmdl
+**/*.jfm
+**/azds.yaml
+**/bin
+**/charts
+**/docker-compose*
+**/Dockerfile*
+**/node_modules
+**/npm-debug.log
+**/obj
+**/secrets.dev.yaml
+**/values.dev.yaml
+LICENSE
+README.md

src/JT808.DotNetty.Abstractions/JT808.DotNetty.Abstractions.csproj

@@ -20,7 +20,7 @@
     <PackageReleaseNotes>基于DotNetty实现的JT808DotNetty的抽象库</PackageReleaseNotes>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="JT808" Version="2.2.1" />
+    <PackageReference Include="JT808" Version="2.2.3" />
   </ItemGroup>
   <ItemGroup>
     <None Include="..\..\LICENSE" Pack="true" PackagePath="" />

src/JT808.DotNetty.CleintBenchmark/Controllers/ReportController.cs

@@ -0,0 +1,63 @@
+using System;
+using System.Collections.Generic;
+using System.ComponentModel.DataAnnotations;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using JT808.DotNetty.Client;
+using JT808.DotNetty.Client.Metadata;
+using JT808.DotNetty.Client.Services;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Cors;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Primitives;
+
+namespace JT808.DotNetty.CleintBenchmark
+{
+    /// <summary>
+    /// 车辆控制器
+    /// </summary>
+    [Route("JT808WebApi")]
+    [ApiController]
+    [EnableCors("Domain")]
+    public class ReportController : ControllerBase
+    {
+        private readonly IJT808TcpClientFactory clientFactory;
+        private readonly JT808ReceiveAtomicCounterService ReceiveAtomicCounterService;
+        private readonly JT808SendAtomicCounterService SendAtomicCounterService;
+
+        /// <summary>
+        /// 
+        /// </summary>
+        public ReportController(
+            IJT808TcpClientFactory factory,
+            JT808ReceiveAtomicCounterService jT808ReceiveAtomicCounterService,
+            JT808SendAtomicCounterService jT808SendAtomicCounterService)
+        {
+            clientFactory = factory;
+            ReceiveAtomicCounterService = jT808ReceiveAtomicCounterService;
+            SendAtomicCounterService = jT808SendAtomicCounterService;
+        }
+
+        [HttpPost]
+        [HttpGet]
+        [Route("QueryReport")]
+        public ActionResult<JT808Report> QueryReport()
+        {
+            var clients = clientFactory.GetAll();
+            JT808Report report = new JT808Report()
+            {
+                SendTotalCount = SendAtomicCounterService.MsgSuccessCount,
+                ReceiveTotalCount = ReceiveAtomicCounterService.MsgSuccessCount,
+                CurrentDate = DateTime.Now,
+                Connections = clients.Count,
+                OnlineConnections = clients.Where(w => w.IsOpen).Count(),
+                OfflineConnections = clients.Where(w => !w.IsOpen).Count(),
+            };
+            return report;
+        }
+    }
+}

src/JT808.DotNetty.CleintBenchmark/JT808.DotNetty.CleintBenchmark.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk.Web">
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
@@ -6,6 +6,11 @@
     <LangVersion>7.3</LangVersion>
   </PropertyGroup>
 
+  <ItemGroup>
+    <Content Remove="wwwroot\echarts.min.js" />
+    <Content Remove="wwwroot\index.html" />
+  </ItemGroup>
+
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Configuration.FileExtensions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.1.0" />
@@ -18,6 +23,21 @@
     <ProjectReference Include="..\JT808.DotNetty.Client\JT808.DotNetty.Client.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <None Include="wwwroot\echarts.min.js">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Include="wwwroot\index.html">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
+  <ItemGroup>
+    <Content Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
+
   <ItemGroup>
     <None Update="appsettings.json">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>

src/JT808.DotNetty.CleintBenchmark/Program.cs

@@ -7,8 +7,10 @@ using System.Threading.Tasks;
 using JT808.Protocol;
 using JT808.DotNetty.Client;
 using JT808.DotNetty.CleintBenchmark.Configs;
-using JT808.DotNetty.CleintBenchmark.Services;
 using Microsoft.Extensions.Configuration;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Builder;
+using JT808.DotNetty.CleintBenchmark.Services;
 
 namespace JT808.DotNetty.CleintBenchmark
 {
@@ -22,6 +24,31 @@ namespace JT808.DotNetty.CleintBenchmark
                 config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
                 config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true);
             })
+            .ConfigureWebHostDefaults(webBuilder => 
+            {
+                webBuilder.Configure(app =>
+                {
+                    app.UseRouting();
+                    app.UseCors("Domain");
+                    app.UseStaticFiles();
+                    app.UseDefaultFiles("/index.html");
+                    app.UseEndpoints(endpoints =>
+                    {
+                        endpoints.MapControllers();
+                    });
+                })
+                .ConfigureServices((hostContext, services) => 
+                {
+                    services.AddControllers();
+                    services.AddCors(options =>
+                        options.AddPolicy("Domain", builder =>
+                        builder.AllowAnyOrigin()
+                               .AllowAnyMethod()
+                               .AllowAnyHeader()
+                               .AllowAnyOrigin()));
+
+                });
+            })
             .ConfigureLogging((context, logging) =>
             {
                 if (Environment.OSVersion.Platform == PlatformID.Unix)

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

@@ -71,7 +71,7 @@ namespace JT808.DotNetty.CleintBenchmark.Services
                         }
                         catch (Exception ex)
                         {
-                            logger.LogError(ex, "");
+                            logger.LogError(ex.Message);
                         }
                     });
                     Thread.Sleep(clientBenchmarkOptions.Interval);

src/JT808.DotNetty.CleintBenchmark/appsettings.json

@@ -3,20 +3,22 @@
     "IncludeScopes": false,
     "Debug": {
       "LogLevel": {
-        "Default": "Trace"
+        "Default": "Error"
       }
     },
     "Console": {
       "LogLevel": {
-        "Default": "Trace"
+        "Default": "Error"
       }
     }
   },
+  "AllowedHosts": "*",
+  //"urls": "http://*:15004;",
   "ClientBenchmarkOptions": {
     "IP": "127.0.0.1",
     "Port": 808,
     "DeviceCount": 100,
     "Interval": 1000,
-    "DeviceTemplate": 300000 //需要多台机器同时访问，那么可以根据这个避开重复终端号 100000-200000-300000
+    "DeviceTemplate": 100000 //需要多台机器同时访问，那么可以根据这个避开重复终端号 100000-200000-300000
   }
 }

src/JT808.DotNetty.CleintBenchmark/wwwroot/index.html

@@ -0,0 +1,153 @@
+<!DOCTYPE html>
+<html>
+
+<head>
+    <meta charset="utf-8" />
+    <title>收发查看</title>
+    <script src="https://unpkg.com/dayjs"></script>
+    <script src="https://unpkg.com/axios/dist/axios.min.js"></script>
+    <script src="/echarts.min.js"></script>
+</head>
+
+<body>
+    <div id="tcpContainer" style="text-align:center;margin:20px auto;width:80%;height: 300px;"></div>
+    <div id="connContainer" style="text-align:center;margin:20px auto;width:80%;height: 300px;"></div>
+    <script type="text/javascript">
+        var tcpDom = document.getElementById("tcpContainer");
+        var connDom = document.getElementById("connContainer");
+        var tcpChart = echarts.init(tcpDom);
+        var connChart = echarts.init(connDom);
+        var sendData=[];
+        var receiveData=[];
+        var onlineData=[];
+        var offlineData=[];
+        var timeData=[];
+        var tcpOption = {
+            title: {
+                text: 'TCP收发数'
+            },
+            tooltip: {
+                trigger: 'axis',
+                axisPointer: {
+                    animation: true
+                }
+            },
+            legend: {
+                data:['发送总次数','接收总次数']
+            },
+            xAxis: {
+                type: 'category',
+                boundaryGap: false,
+                data: timeData
+            },
+            yAxis: {
+                type: 'value',
+                boundaryGap: [0, '100%'],
+                splitLine: {
+                    show: true
+                }
+            },
+            series: [{
+                name: '发送总次数',
+                type: 'line',
+                color: "blue",
+                data: sendData
+            },{
+                name: '接收总次数',
+                type: 'line',
+                color: "red",
+                data: receiveData
+            }]
+        };
+        var connOption = {
+            title: {
+                text: 'TCP连接数'
+            },
+            tooltip: {
+                trigger: 'axis',
+                axisPointer: {
+                    animation: true
+                }
+            },
+            legend: {
+                data:['tcp在线数','tcp离线数']
+            },
+            xAxis: {
+                type: 'category', 
+                boundaryGap: false,
+                data: timeData
+            },
+            yAxis: {
+                type: 'value',
+                boundaryGap: [0, '100%'],
+                splitLine: {
+                    show: true
+                }
+            },
+            series: [{
+                name: 'tcp在线数',
+                type: 'line',
+                color: "blue",
+                data: onlineData
+            },{
+                name: 'tcp离线数',
+                type: 'line',
+                color: "red",
+                data: offlineData
+            }]
+        };
+        setInterval(function () {
+            axios.post('http://localhost:5000/JT808WebApi/QueryReport')
+            .then((response) => {
+                    if (response.data) {
+                        if(sendData.length>16){
+                            sendData.shift();
+                            receiveData.shift();
+                            onlineData.shift();
+                            offlineData.shift();
+                            timeData.shift();
+                        }
+                        //console.log(response.data);
+                        timeData.push(dayjs(response.data.currentDate).format('HH:mm:ss'));
+                        sendData.push(response.data.sendTotalCount);
+                        receiveData.push(response.data.receiveTotalCount);
+                        onlineData.push(response.data.onlineConnections);
+                        offlineData.push(response.data.offlineConnections);
+                        tcpChart.setOption({
+                            series: [{
+                                data: sendData
+                            },{
+                                data: receiveData
+                            }],
+                            xAxis:[{
+                                data: timeData
+                            }]
+                        });
+                        connChart.setOption({
+                            series: [{
+                                data: onlineData
+                            },{
+                                data: offlineData
+                            }],
+                            xAxis:[{
+                                data: timeData
+                            }]
+                        });
+                    } else {
+                        alert("没有数据");
+                    }
+                })
+                .catch((error) => {
+                    console.log(error); 
+                });
+        }, 1000);
+        if (tcpOption && typeof tcpOption === "object") {
+            tcpChart.setOption(tcpOption, true);
+        }
+        if (connOption && typeof connOption === "object") {
+            connChart.setOption(connOption, true);
+        }
+    </script>
+</body>
+
+</html>

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

@@ -24,7 +24,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.2.1" />
+    <PackageReference Include="JT808" Version="2.2.3" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.1.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />

src/JT808.DotNetty.Client/JT808ClientMsgSNDistributedImpl.cs

@@ -1,16 +0,0 @@
-using JT808.Protocol;
-using JT808.Protocol.Interfaces;
-using System.Threading;
-
-namespace JT808.DotNetty.Client
-{
-    internal class JT808ClientMsgSNDistributedImpl : IJT808MsgSNDistributed
-    {
-        int _counter = 0;
-
-        public ushort Increment()
-        {
-            return (ushort)Interlocked.Increment(ref _counter);
-        }
-    }
-}

src/JT808.DotNetty.Client/JT808DeviceConfig.cs

@@ -14,7 +14,6 @@ namespace JT808.DotNetty.Client
             TerminalPhoneNo = terminalPhoneNo;
             TcpHost = tcpHost;
             TcpPort = tcpPort;
-            MsgSNDistributed = new JT808ClientMsgSNDistributedImpl();
             Version = version;
         }
         public JT808Version Version { get; private set; }

src/JT808.DotNetty.Client/JT808TcpClient.cs

@@ -64,14 +64,14 @@ namespace JT808.DotNetty.Client
             });
         }
 
-        public async void Send(JT808ClientRequest request)
+        public  void Send(JT808ClientRequest request)
         {
             if (disposed) return;
             if (clientChannel == null) throw new NullReferenceException("Channel is empty.");
             if (request == null) throw new ArgumentNullException("JT808ClientRequest Parameter is empty.");
             if (clientChannel.Active && clientChannel.Open)
             {
-                await clientChannel.WriteAndFlushAsync(request);
+                 clientChannel.WriteAndFlushAsync(request);
             }
         }
 

src/JT808.DotNetty.Client/JT808TcpClientExtensions.cs

@@ -14,7 +14,6 @@ namespace JT808.DotNetty.Client
         public static void Send(this JT808TcpClient client, JT808Package package, int minBufferSize = 4096)
         {
             package.Header.TerminalPhoneNo = client.DeviceConfig.TerminalPhoneNo;
-            package.Header.MsgNum = client.DeviceConfig.MsgSNDistributed.Increment(); 
             JT808ClientRequest request = new JT808ClientRequest(package, minBufferSize);
             client.Send(request);
         }

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

@@ -294,11 +294,11 @@ namespace JT808.DotNetty.Core.Session
         }
         public IEnumerable<JT808TcpSession> GetTcpAll()
         {
-            return Sessions.Select(s => (JT808TcpSession)s.Value).Where(w => w.TransportProtocolType == JT808TransportProtocolType.tcp).ToList();
+            return Sessions.Where(w => w.Value.TransportProtocolType == JT808TransportProtocolType.tcp).Select(s => (JT808TcpSession)s.Value).ToList();
         }
         public IEnumerable<JT808UdpSession> GetUdpAll()
         {
-            return Sessions.Select(s => (JT808UdpSession)s.Value).Where(w => w.TransportProtocolType == JT808TransportProtocolType.udp).ToList();
+            return Sessions.Where(w => w.Value.TransportProtocolType == JT808TransportProtocolType.udp).Select(s => (JT808UdpSession)s.Value).ToList();
         }
     }
 }

src/JT808.DotNetty.Services/JT808.DotNetty.Traffic/JT808.DotNetty.Traffic.csproj

@@ -24,7 +24,7 @@
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.1.0" />
-    <PackageReference Include="CSRedisCore" Version="3.2.0" />
+    <PackageReference Include="CSRedisCore" Version="3.2.1" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\JT808.DotNetty.Abstractions\JT808.DotNetty.Abstractions.csproj" />

src/JT808.Gateway/Enums/JT808TransportProtocolType.cs → src/JT808.Gateway.Abstractions/Enums/JT808TransportProtocolType.cs

@@ -2,7 +2,7 @@
 using System.Collections.Generic;
 using System.Text;
 
-namespace JT808.Gateway.Enums
+namespace JT808.Gateway.Abstractions.Enums
 {
     /// <summary>
     /// 传输协议类型

src/JT808.Gateway/IJT808ClientBuilder.cs → src/JT808.Gateway.Abstractions/IJT808ClientBuilder.cs

@@ -4,7 +4,7 @@ using System;
 using System.Collections.Generic;
 using System.Text;
 
-namespace JT808.Gateway
+namespace JT808.Gateway.Abstractions
 {
     public interface IJT808ClientBuilder
     {

src/JT808.Gateway/IJT808GatewayBuilder.cs → src/JT808.Gateway.Abstractions/IJT808GatewayBuilder.cs

@@ -4,7 +4,7 @@ using System;
 using System.Collections.Generic;
 using System.Text;
 
-namespace JT808.Gateway
+namespace JT808.Gateway.Abstractions
 {
     public interface IJT808GatewayBuilder
     {

src/JT808.Gateway/PubSub/IJT808MsgConsumer.cs → src/JT808.Gateway.Abstractions/IJT808MsgConsumer.cs

@@ -3,7 +3,7 @@ using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 
-namespace JT808.Gateway.PubSub
+namespace JT808.Gateway.Abstractions
 {
     public interface IJT808MsgConsumer : IJT808PubSub, IDisposable
     {

src/JT808.Gateway/PubSub/IJT808MsgProducer.cs → src/JT808.Gateway.Abstractions/IJT808MsgProducer.cs

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

src/JT808.Gateway/PubSub/IJT808MsgReplyConsumer.cs → src/JT808.Gateway.Abstractions/IJT808MsgReplyConsumer.cs

@@ -3,7 +3,7 @@ using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 
-namespace JT808.Gateway.PubSub
+namespace JT808.Gateway.Abstractions
 {
     public interface IJT808MsgReplyConsumer : IJT808PubSub, IDisposable
     {

src/JT808.Gateway/PubSub/IJT808MsgReplyProducer.cs → src/JT808.Gateway.Abstractions/IJT808MsgReplyProducer.cs

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

src/JT808.Gateway/PubSub/IJT808PubSub.cs → src/JT808.Gateway.Abstractions/IJT808PubSub.cs

@@ -2,7 +2,7 @@
 using System.Collections.Generic;
 using System.Text;
 
-namespace JT808.Gateway.PubSub
+namespace JT808.Gateway.Abstractions
 {
     public interface IJT808PubSub
     {

src/JT808.Gateway/PubSub/IJT808SessionConsumer.cs → src/JT808.Gateway.Abstractions/IJT808SessionConsumer.cs

@@ -3,7 +3,7 @@ using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 
-namespace JT808.Gateway.PubSub
+namespace JT808.Gateway.Abstractions
 {
     /// <summary>
     /// 会话通知（在线/离线）

src/JT808.Gateway/PubSub/IJT808SessionProducer.cs → src/JT808.Gateway.Abstractions/IJT808SessionProducer.cs

@@ -1,13 +1,13 @@
 using System;
 using System.Threading.Tasks;
 
-namespace JT808.Gateway.PubSub
+namespace JT808.Gateway.Abstractions
 {
     /// <summary>
     /// 会话通知（在线/离线）
     /// </summary>
     public interface IJT808SessionProducer : IJT808PubSub, IDisposable
     {
-        Task ProduceAsync(string notice,string terminalNo);
+        ValueTask ProduceAsync(string notice,string terminalNo);
     }
 }

src/JT808.Gateway.Abstractions/JT808.Gateway.Abstractions.csproj

@@ -0,0 +1,33 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>netstandard2.1</TargetFramework>
+    <LangVersion>8.0</LangVersion>
+    <Copyright>Copyright 2019.</Copyright>
+    <Authors>SmallChi(Koike)</Authors>
+    <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <PackageLicenseFile>LICENSE</PackageLicenseFile>
+    <PackageRequireLicenseAcceptance>true</PackageRequireLicenseAcceptance>
+    <Description>基于Pipeline实现的JT808Gateway的抽象库</Description>
+    <PackageReleaseNotes>基于Pipeline实现的JT808Gateway的抽象库</PackageReleaseNotes>
+    <PackageId>JT808.Gateway.Abstractions</PackageId>
+    <Product>JT808.Gateway.Abstractions</Product>
+    <Version>1.0.0-preview2</Version>
+  </PropertyGroup>
+  <ItemGroup>
+    <Protobuf Include="Protos\JT808Gateway.proto" />
+  </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="Google.Protobuf" Version="3.11.2" />
+    <PackageReference Include="Grpc.Core" Version="2.25.0" />
+    <PackageReference Include="Grpc.Tools" Version="2.25.0">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="JT808" Version="2.2.2" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.0" />
+  </ItemGroup>
+  <ItemGroup>
+    <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
+  </ItemGroup>
+</Project>

src/JT808.Gateway.Abstractions/JT808GatewayConstants.cs

@@ -0,0 +1,11 @@
+namespace JT808.Gateway.Abstractions
+{
+    public static class JT808GatewayConstants
+    {
+        public const string SessionOnline= "JT808SessionOnline";
+        public const string SessionOffline = "JT808SessionOffline";
+        public const string SessionTopic = "jt808session";
+        public const string MsgTopic = "jt808msgdefault";
+        public const string MsgReplyTopic = "jt808msgreplydefault";
+    }
+}

src/JT808.Gateway.CleintBenchmark/Configs/ClientBenchmarkOptions.cs

@@ -1,19 +0,0 @@
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.CleintBenchmark.Configs
-{
-    public class ClientBenchmarkOptions : IOptions<ClientBenchmarkOptions>
-    {
-        public string IP { get; set; }
-        public int Port { get; set; }
-        public int DeviceCount { get; set; } = 10;
-        /// <summary>
-        /// 5000ms毫秒
-        /// </summary>
-        public int Interval { get; set; } = 5000;
-        public ClientBenchmarkOptions Value =>this;
-    }
-}

src/JT808.Gateway.CleintBenchmark/Configs/nlog.win.config

@@ -1,35 +0,0 @@
-<?xml version="1.0" encoding="utf-8" ?>
-  <!--
-      参考：http://www.cnblogs.com/fuchongjundream/p/3936431.html
-      autoReload：自动再配置
-      internalLogFile：可以让NLog把内部的调试和异常信息都写入指定文件里程序没问题了，日志却出了问题。这个该怎么办，到底是哪里不正确了？假如日志本身除了bug该如何解决？这就需要日志排错。把日志的错误信息写入日志。
-      <nlog throwExceptions="true" />    
-      <nlog internalLogFile="file.txt" />- 设置internalLogFile属性可以让NLog把内部的调试和异常信息都写入指定文件里。
-      <nlog internalLogLevel="Trace|Debug|Info|Warn|Error|Fatal" /> - 决定内部日志的级别，级别越高，输出的日志信息越简洁。
-      <nlog internalLogToConsole="false|true" /> - 是否把内部日志输出到标准控制台。
-      <nlog internalLogToConsoleError="false|true" /> - 是否把内部日志输出到标准错误控制台 (stderr)。
-      设置throwExceptions属性为“true”可以让NLog不再阻挡这类异常，而是把它们抛给调用者。在部署是这样做可以帮我们快速定位问题。一旦应用程序已经正确配置了，我们建议把throwExceptions的值设为“false”，这样由于日志引发的问题不至于导致应用程序的崩溃。
-  -->
-  <nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xsi:schemaLocation="NLog NLog.xsd"
-      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-      autoReload="true"
-      internalLogFile="${basedir}/wwwroot/logs/internalLog.txt"
-      internalLogLevel="Debug" >
-    <targets>
-      <target name="all" xsi:type="File"
-        fileName="${basedir}/wwwroot/logs/all/${shortdate}.log"
-        layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level}:${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}"/>
-      <target name="console" xsi:type="ColoredConsole"
-             useDefaultRowHighlightingRules="false"
-             layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level} ${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}">
-        <highlight-row condition="level == LogLevel.Debug" foregroundColor="DarkGray" />
-        <highlight-row condition="level == LogLevel.Info" foregroundColor="Gray" />
-        <highlight-row condition="level == LogLevel.Warn" foregroundColor="Yellow" />
-        <highlight-row condition="level == LogLevel.Error" foregroundColor="Red" />
-        <highlight-row condition="level == LogLevel.Fatal" foregroundColor="Red" backgroundColor="White" />
-      </target>
-    </targets>
-    <rules>
-      <logger name="*" minlevel="Debug" maxlevel="Fatal" writeTo="all,console"/>
-    </rules>
-  </nlog>

src/JT808.Gateway.CleintBenchmark/Program.cs

@@ -1,51 +0,0 @@
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using NLog.Extensions.Logging;
-using System;
-using System.Threading.Tasks;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using JT808.Gateway.CleintBenchmark.Configs;
-using JT808.Gateway.Client;
-using JT808.Gateway.CleintBenchmark.Services;
-
-namespace JT808.Gateway.CleintBenchmark
-{
-    class Program
-    {
-        static async Task Main(string[] args)
-        {
-            var serverHostBuilder = new HostBuilder()
-            .ConfigureAppConfiguration((hostingContext, config) =>
-            {
-                config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true);
-            })
-            .ConfigureLogging((context, logging) =>
-            {
-                if (Environment.OSVersion.Platform == PlatformID.Unix)
-                {
-                    NLog.LogManager.LoadConfiguration("Configs/nlog.unix.config");
-                }
-                else
-                {
-                    NLog.LogManager.LoadConfiguration("Configs/nlog.win.config");
-                }
-                logging.AddNLog();
-                logging.SetMinimumLevel(LogLevel.Trace);
-            })
-            .ConfigureServices((hostContext, services) =>
-            {
-                services.Configure<ClientBenchmarkOptions>(hostContext.Configuration.GetSection("ClientBenchmarkOptions"));
-                services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                services.AddJT808Configure()
-                        .AddJT808Client();
-                services.AddHostedService<CleintBenchmarkHostedService>();
-                services.AddHostedService<CleintBenchmarkReportHostedService>();
-            });
-            await serverHostBuilder.RunConsoleAsync();
-        }
-    }
-}

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

@@ -1,82 +0,0 @@
-using JT808.Gateway.CleintBenchmark.Configs;
-using JT808.Gateway.Client;
-using JT808.Protocol.MessageBody;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-
-namespace JT808.Gateway.CleintBenchmark.Services
-{
-    public class CleintBenchmarkHostedService : IHostedService
-    {
-        private readonly ClientBenchmarkOptions clientBenchmarkOptions;
-
-        private readonly ILogger logger;
-
-        private readonly IJT808TcpClientFactory jT808TcpClientFactory;
-
-        private CancellationTokenSource cts=new CancellationTokenSource();
-
-        private TaskFactory taskFactory;
-
-        public CleintBenchmarkHostedService(
-            ILoggerFactory loggerFactory,
-            IJT808TcpClientFactory jT808TcpClientFactory,
-            IOptions<ClientBenchmarkOptions> clientBenchmarkOptionsAccessor)
-        {
-            this.jT808TcpClientFactory = jT808TcpClientFactory;
-            clientBenchmarkOptions = clientBenchmarkOptionsAccessor.Value;
-            logger = loggerFactory.CreateLogger("CleintBenchmarkHostedService");
-            taskFactory = new TaskFactory();
-        }
-        public Task StartAsync(CancellationToken cancellationToken)
-        {
-            logger.LogInformation("StartAsync...");
-            ThreadPool.GetMinThreads(out var minWorkerThreads, out var minCompletionPortThreads);
-            ThreadPool.GetMaxThreads(out var maxWorkerThreads, out var maxCompletionPortThreads);
-            logger.LogInformation($"GetMinThreads:{minWorkerThreads}-{minCompletionPortThreads}");
-            logger.LogInformation($"GetMaxThreads:{maxWorkerThreads}-{maxCompletionPortThreads}");
-            //ThreadPool.SetMaxThreads(20, 20);
-            //ThreadPool.GetMaxThreads(out var setMaxWorkerThreads, out var setMaxCompletionPortThreads);
-            //logger.LogInformation($"SetMaxThreads:{setMaxWorkerThreads}-{setMaxCompletionPortThreads}");
-            for (int i=0;i< clientBenchmarkOptions.DeviceCount; i++)
-            {
-                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);
-                    while (!cts.IsCancellationRequested)
-                    {
-                        client.Send(new JT808_0x0200()
-                        {
-                            Lat = lat,
-                            Lng = Lng,
-                            GPSTime = DateTime.Now,
-                            Speed = 50,
-                            Direction = 30,
-                            AlarmFlag = 5,
-                            Altitude = 50,
-                            StatusFlag = 10
-                        });
-                        Thread.Sleep(clientBenchmarkOptions.Interval);
-                    }
-                }, i,cts.Token);
-            }
-            return Task.CompletedTask;
-        }
-
-        public Task StopAsync(CancellationToken cancellationToken)
-        {
-            cts.Cancel();
-            logger.LogInformation("StopAsync...");
-            return Task.CompletedTask;
-        }
-    }
-}

src/JT808.Gateway.CleintBenchmark/Services/CleintBenchmarkReportHostedService.cs

@@ -1,53 +0,0 @@
-using JT808.Gateway.Services;
-using JT808.Protocol.MessageBody;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using Newtonsoft.Json;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-
-namespace JT808.Gateway.CleintBenchmark.Services
-{
-    public class CleintBenchmarkReportHostedService : IHostedService
-    {
-        private readonly JT808ClientReportService jT808ReportService;
-
-        private CancellationTokenSource cts=new CancellationTokenSource();
-
-        private readonly ILogger logger;
-        public CleintBenchmarkReportHostedService(
-            ILoggerFactory loggerFactory,
-            JT808ClientReportService jT808ReportService)
-        {
-            this.jT808ReportService = jT808ReportService;
-            logger = loggerFactory.CreateLogger("CleintBenchmarkReportHostedService");
-        }
-        public Task StartAsync(CancellationToken cancellationToken)
-        {
-            logger.LogInformation("StartAsync...");
-            Task.Run(() => {
-                while (!cts.IsCancellationRequested)
-                {
-                    logger.LogInformation(JsonConvert.SerializeObject(jT808ReportService.JT808Reports.LastOrDefault()));
-                    Thread.Sleep(3000);
-                }
-            }, cts.Token);
-            return Task.CompletedTask;
-        }
-
-        public Task StopAsync(CancellationToken cancellationToken)
-        {
-            logger.LogInformation("StopAsync...");
-            cts.Cancel();
-            logger.LogInformation("正在生成报表...");
-            logger.LogInformation(JsonConvert.SerializeObject(jT808ReportService.JT808Reports,Formatting.Indented));
-            return Task.CompletedTask;
-        }
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808ConsumerConfig.cs

@@ -1,15 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808ConsumerConfig: ConsumerConfig, IOptions<JT808ConsumerConfig>
-    {
-        public string TopicName { get; set; }
-
-        public JT808ConsumerConfig Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808MsgConsumerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808MsgConsumerConfig : JT808ConsumerConfig, IOptions<JT808MsgConsumerConfig>
-    {
-        JT808MsgConsumerConfig IOptions<JT808MsgConsumerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808MsgProducerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808MsgProducerConfig : JT808ProducerConfig, IOptions<JT808MsgProducerConfig>
-    {
-        JT808MsgProducerConfig IOptions<JT808MsgProducerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808MsgReplyConsumerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808MsgReplyConsumerConfig : JT808ConsumerConfig, IOptions<JT808MsgReplyConsumerConfig>
-    {
-        JT808MsgReplyConsumerConfig IOptions<JT808MsgReplyConsumerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808MsgReplyProducerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808MsgReplyProducerConfig : JT808ProducerConfig, IOptions<JT808MsgReplyProducerConfig>
-    {
-        JT808MsgReplyProducerConfig IOptions<JT808MsgReplyProducerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808ProducerConfig.cs

@@ -1,15 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808ProducerConfig : ProducerConfig,IOptions<JT808ProducerConfig>
-    {
-        public string TopicName { get; set; }
-
-        public JT808ProducerConfig Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808SessionConsumerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808SessionConsumerConfig : JT808ConsumerConfig, IOptions<JT808SessionConsumerConfig>
-    {
-        JT808SessionConsumerConfig IOptions<JT808SessionConsumerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/Configs/JT808SessionProducerConfig.cs

@@ -1,13 +0,0 @@
-using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Configs.Kafka
-{
-    public class JT808SessionProducerConfig : JT808ProducerConfig, IOptions<JT808SessionProducerConfig>
-    {
-        JT808SessionProducerConfig IOptions<JT808SessionProducerConfig>.Value => this;
-    }
-}

src/JT808.Gateway.Kafka/JT808.Gateway.Kafka.csproj

@@ -1,39 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
-    <LangVersion>8.0</LangVersion>
-    <Copyright>Copyright 2018.</Copyright>
-    <Authors>SmallChi(Koike)</Authors>
-    <RepositoryUrl>https://github.com/SmallChi/JT808DotNetty</RepositoryUrl>
-    <PackageProjectUrl>https://github.com/SmallChi/JT808DotNetty</PackageProjectUrl>
-    <licenseUrl>https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE</licenseUrl>
-    <license>https://github.com/SmallChi/JT808DotNetty/blob/master/LICENSE</license>
-    <GeneratePackageOnBuild>false</GeneratePackageOnBuild>
-    <Version>1.0.0-preview1</Version>
-    <SignAssembly>false</SignAssembly>
-    <PackageLicenseFile>LICENSE</PackageLicenseFile>
-    <PackageRequireLicenseAcceptance>true</PackageRequireLicenseAcceptance>
-    <PackageId>JT808.Gateway.Kafka</PackageId>
-    <Product>JT808.Gateway.Kafka</Product>
-    <Description>基于Kafka的JT808消息发布与订阅</Description>
-    <PackageReleaseNotes>基于Kafka的JT808消息发布与订阅</PackageReleaseNotes>
-  </PropertyGroup>
-  <ItemGroup>
-    <PackageReference Include="Confluent.Kafka" Version="1.2.1" />
-    <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Options" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.0.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.Kafka/JT808ClientBuilderDefault.cs

@@ -1,24 +0,0 @@
-using JT808.Protocol;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace JT808.Gateway.Kafka
-{
-    internal class JT808ClientBuilderDefault : IJT808ClientBuilder
-    {
-        public IJT808Builder JT808Builder { get; }
-
-        public JT808ClientBuilderDefault(IJT808Builder builder)
-        {
-            JT808Builder = builder;
-        }
-
-        public IJT808Builder Builder()
-        {
-            return JT808Builder;
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808ClientKafkaExtensions.cs

@@ -1,66 +0,0 @@
-using JJT808.Gateway.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-
-namespace JT808.Gateway.Kafka
-{
-    public static class JT808ClientKafkaExtensions
-    {
-        public static IJT808ClientBuilder AddJT808ClientKafka(this IJT808Builder builder)
-        {
-            return new JT808ClientBuilderDefault(builder);
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="serviceDescriptors"></param>
-        /// <param name="configuration">GetSection("JT808MsgConsumerConfig")</param>
-        /// <returns></returns>
-        public static IJT808ClientBuilder AddMsgConsumer(this IJT808ClientBuilder jT808ClientBuilder, IConfiguration configuration)
-        {
-            jT808ClientBuilder.JT808Builder.Services.Configure<JT808MsgConsumerConfig>(configuration.GetSection("JT808MsgConsumerConfig"));
-            jT808ClientBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgConsumer, JT808MsgConsumer>();
-            return jT808ClientBuilder;
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="serviceDescriptors"></param>
-        /// <param name="configuration">GetSection("JT808MsgReplyProducerConfig")</param>
-        /// <returns></returns>
-        public static IJT808ClientBuilder AddMsgReplyProducer(this IJT808ClientBuilder jT808ClientBuilder, IConfiguration configuration)
-        {
-            jT808ClientBuilder.JT808Builder.Services.Configure<JT808MsgReplyProducerConfig>(configuration.GetSection("JT808MsgReplyProducerConfig"));
-            jT808ClientBuilder.JT808Builder.Services.TryAddSingleton<IJT808MsgReplyProducer, JT808MsgReplyProducer>();
-            return jT808ClientBuilder;
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="jT808NettyBuilder"></param>
-        /// <param name="configuration">GetSection("JT808MsgReplyConsumerConfig")</param>
-        /// <returns></returns>
-        public static IJT808ClientBuilder AddMsgReplyConsumer(this IJT808ClientBuilder jT808ClientBuilder, IConfiguration configuration)
-        {
-            jT808ClientBuilder.JT808Builder.Services.Configure<JT808MsgReplyConsumerConfig>(configuration.GetSection("JT808MsgReplyConsumerConfig"));
-            jT808ClientBuilder.JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808MsgReplyConsumer), typeof(JT808MsgReplyConsumer), ServiceLifetime.Singleton));
-            return jT808ClientBuilder;
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="serviceDescriptors"></param>
-        /// <param name="configuration">GetSection("JT808SessionConsumerConfig")</param>
-        /// <returns></returns>
-        public static IJT808ClientBuilder AddSessionConsumer(this IJT808ClientBuilder jT808ClientBuilder, IConfiguration configuration)
-        {
-            jT808ClientBuilder.JT808Builder.Services.Configure<JT808SessionConsumerConfig>(configuration.GetSection("JT808SessionConsumerConfig"));
-            jT808ClientBuilder.JT808Builder.Services.TryAddSingleton<IJT808SessionConsumer, JT808SessionConsumer>();
-            return jT808ClientBuilder;
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808MsgConsumer.cs

@@ -1,82 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.Kafka
-{
-    public class JT808MsgConsumer : IJT808MsgConsumer
-    {
-        public CancellationTokenSource Cts => new CancellationTokenSource();
-
-        private readonly IConsumer<string, byte[]> consumer;
-
-        private readonly ILogger logger;
-
-        public string TopicName { get; }
-
-        public JT808MsgConsumer(
-            IOptions<JT808MsgConsumerConfig> consumerConfigAccessor,
-            ILoggerFactory loggerFactory)
-        {
-            consumer = new ConsumerBuilder<string, byte[]>(consumerConfigAccessor.Value).Build();
-            TopicName = consumerConfigAccessor.Value.TopicName;
-            logger = loggerFactory.CreateLogger("JT808MsgConsumer");
-        }
-
-        public void OnMessage(Action<(string TerminalNo, byte[] Data)> callback)
-        {
-            Task.Run(() =>
-            {
-                while (!Cts.IsCancellationRequested)
-                {
-                    try
-                    {
-                        //如果不指定分区，根据kafka的机制会从多个分区中拉取数据
-                        //如果指定分区，根据kafka的机制会从相应的分区中拉取数据
-                        var data = consumer.Consume(Cts.Token);
-                        if (logger.IsEnabled(LogLevel.Debug))
-                        {
-                            logger.LogDebug($"Topic: {data.Topic} Key: {data.Key} Partition: {data.Partition} Offset: {data.Offset} TopicPartitionOffset:{data.TopicPartitionOffset}");
-                        }
-                        callback((data.Key, data.Value));
-                    }
-                    catch (ConsumeException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (OperationCanceledException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (Exception ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                }
-            }, Cts.Token);
-        }
-
-        public void Subscribe()
-        {
-            consumer.Subscribe(TopicName);
-        }
-
-        public void Unsubscribe()
-        {
-            consumer.Unsubscribe();
-        }
-
-        public void Dispose()
-        {
-            consumer.Close();
-            consumer.Dispose();
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808MsgProducer.cs

@@ -1,38 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.Kafka
-{
-    public class JT808MsgProducer : IJT808MsgProducer
-    {
-        public string TopicName { get; }
-
-        private readonly IProducer<string, byte[]> producer;
-        public JT808MsgProducer(
-          IOptions<JT808MsgProducerConfig> producerConfigAccessor)
-        {
-            producer = new ProducerBuilder<string, byte[]>(producerConfigAccessor.Value).Build();
-            TopicName = producerConfigAccessor.Value.TopicName;
-        }
-
-        public void Dispose()
-        {
-            producer.Dispose();
-        }
-
-        public async Task ProduceAsync(string terminalNo, byte[] data)
-        {
-            await producer.ProduceAsync(TopicName, new Message<string, byte[]>
-            {
-                Key = terminalNo,
-                Value = data
-            });
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808MsgReplyConsumer.cs

@@ -1,82 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.Kafka
-{
-    public class JT808MsgReplyConsumer : IJT808MsgReplyConsumer
-    {
-        public CancellationTokenSource Cts => new CancellationTokenSource();
-
-        private readonly IConsumer<string, byte[]> consumer;
-
-        private readonly ILogger logger;
-
-        public string TopicName { get; }
-
-        public JT808MsgReplyConsumer(
-            IOptions<JT808MsgReplyConsumerConfig> consumerConfigAccessor,
-            ILoggerFactory loggerFactory)
-        {
-            consumer = new ConsumerBuilder<string, byte[]>(consumerConfigAccessor.Value).Build();
-            TopicName = consumerConfigAccessor.Value.TopicName;
-            logger = loggerFactory.CreateLogger("JT808MsgReplyConsumer");
-        }
-
-        public void OnMessage(Action<(string TerminalNo, byte[] Data)> callback)
-        {
-            Task.Run(() =>
-            {
-                while (!Cts.IsCancellationRequested)
-                {
-                    try
-                    {
-                        //如果不指定分区，根据kafka的机制会从多个分区中拉取数据
-                        //如果指定分区，根据kafka的机制会从相应的分区中拉取数据
-                        var data = consumer.Consume(Cts.Token);
-                        if (logger.IsEnabled(LogLevel.Debug))
-                        {
-                            logger.LogDebug($"Topic: {data.Topic} Key: {data.Key} Partition: {data.Partition} Offset: {data.Offset} TopicPartitionOffset:{data.TopicPartitionOffset}");
-                        }
-                        callback((data.Key, data.Value));
-                    }
-                    catch (ConsumeException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (OperationCanceledException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (Exception ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                }
-            }, Cts.Token);
-        }
-
-        public void Subscribe()
-        {
-            consumer.Subscribe(TopicName);
-        }
-
-        public void Unsubscribe()
-        {
-            consumer.Unsubscribe();
-        }
-
-        public void Dispose()
-        {
-            consumer.Close();
-            consumer.Dispose();
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808MsgReplyProducer.cs

@@ -1,38 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JJT808.Gateway.Kafka
-{
-    public class JT808MsgReplyProducer : IJT808MsgReplyProducer
-    {
-        public string TopicName { get;}
-
-        private IProducer<string, byte[]> producer;
-        public JT808MsgReplyProducer(
-          IOptions<JT808MsgReplyProducerConfig> producerConfigAccessor)
-        {
-            producer = new ProducerBuilder<string, byte[]>(producerConfigAccessor.Value).Build();
-            TopicName = producerConfigAccessor.Value.TopicName;
-        }
-
-        public void Dispose()
-        {
-            producer.Dispose();
-        }
-
-        public async Task ProduceAsync(string terminalNo, byte[] data)
-        {
-            await producer.ProduceAsync(TopicName, new Message<string, byte[]>
-            {
-                Key = terminalNo,
-                Value = data
-            });
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808ServerKafkaExtensions.cs

@@ -1,48 +0,0 @@
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.DependencyInjection.Extensions;
-
-namespace JT808.Gateway.Kafka
-{
-    public static class JT808ServerKafkaExtensions
-    {
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="jT808NettyBuilder"></param>
-        /// <param name="configuration">GetSection("JT808MsgProducerConfig")</param>
-        /// <returns></returns>
-        public static IJT808GatewayBuilder AddJT808ServerKafkaMsgProducer(this IJT808GatewayBuilder jT808GatewayBuilder, IConfiguration configuration)
-        {
-            jT808GatewayBuilder.JT808Builder.Services.Configure<JT808MsgProducerConfig>(configuration.GetSection("JT808MsgProducerConfig"));
-            jT808GatewayBuilder.JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808MsgProducer), typeof(JT808MsgProducer), ServiceLifetime.Singleton));
-            return jT808GatewayBuilder;
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="jT808NettyBuilder"></param>
-        /// <param name="configuration">GetSection("JT808MsgReplyConsumerConfig")</param>
-        /// <returns></returns>
-        public static IJT808GatewayBuilder AddJT808ServerKafkaMsgReplyConsumer(this IJT808GatewayBuilder jT808GatewayBuilder, IConfiguration configuration)
-        {
-            jT808GatewayBuilder.JT808Builder.Services.Configure<JT808MsgReplyConsumerConfig>(configuration.GetSection("JT808MsgReplyConsumerConfig"));
-            jT808GatewayBuilder.JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808MsgReplyConsumer), typeof(JT808MsgReplyConsumer), ServiceLifetime.Singleton));
-            return jT808GatewayBuilder;
-        }
-        /// <summary>
-        /// 
-        /// </summary>
-        /// <param name="jT808NettyBuilder"></param>
-        /// <param name="configuration">GetSection("JT808SessionProducerConfig")</param>
-        /// <returns></returns>
-        public static IJT808GatewayBuilder AddJT808ServerKafkaSessionProducer(this IJT808GatewayBuilder jT808GatewayBuilder, IConfiguration configuration)
-        {
-            jT808GatewayBuilder.JT808Builder.Services.Configure<JT808SessionProducerConfig>(configuration.GetSection("JT808SessionProducerConfig"));
-            jT808GatewayBuilder.JT808Builder.Services.Replace(new ServiceDescriptor(typeof(IJT808SessionProducer), typeof(JT808SessionProducer), ServiceLifetime.Singleton));
-            return jT808GatewayBuilder;
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808SessionConsumer.cs

@@ -1,82 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.Kafka
-{
-    public class JT808SessionConsumer : IJT808SessionConsumer
-    {
-        public CancellationTokenSource Cts => new CancellationTokenSource();
-
-        private readonly IConsumer<string, string> consumer;
-
-        private readonly ILogger logger;
-
-        public string TopicName { get; }
-
-        public JT808SessionConsumer(
-            IOptions<JT808SessionConsumerConfig> consumerConfigAccessor,
-            ILoggerFactory loggerFactory)
-        {
-            consumer = new ConsumerBuilder<string, string>(consumerConfigAccessor.Value).Build();
-            TopicName = consumerConfigAccessor.Value.TopicName;
-            logger = loggerFactory.CreateLogger("JT808SessionConsumer");
-        }
-
-        public void OnMessage(Action<(string Notice, string TerminalNo)> callback)
-        {
-            Task.Run(() =>
-            {
-                while (!Cts.IsCancellationRequested)
-                {
-                    try
-                    {
-                        //如果不指定分区，根据kafka的机制会从多个分区中拉取数据
-                        //如果指定分区，根据kafka的机制会从相应的分区中拉取数据
-                        var data = consumer.Consume(Cts.Token);
-                        if (logger.IsEnabled(LogLevel.Debug))
-                        {
-                            logger.LogDebug($"Topic: {data.Topic} Key: {data.Key} Partition: {data.Partition} Offset: {data.Offset} TopicPartitionOffset:{data.TopicPartitionOffset}");
-                        }
-                        callback((data.Key, data.Value));
-                    }
-                    catch (ConsumeException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (OperationCanceledException ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                    catch (Exception ex)
-                    {
-                        logger.LogError(ex, TopicName);
-                    }
-                }
-            }, Cts.Token);
-        }
-
-        public void Subscribe()
-        {
-            consumer.Subscribe(TopicName);
-        }
-
-        public void Unsubscribe()
-        {
-            consumer.Unsubscribe();
-        }
-
-        public void Dispose()
-        {
-            consumer.Close();
-            consumer.Dispose();
-        }
-    }
-}

src/JT808.Gateway.Kafka/JT808SessionProducer.cs

@@ -1,38 +0,0 @@
-using Confluent.Kafka;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.Kafka
-{
-    public class JT808SessionProducer : IJT808SessionProducer
-    {
-        public string TopicName { get; }
-
-        private readonly IProducer<string, string> producer;
-        public JT808SessionProducer(
-          IOptions<JT808SessionProducerConfig> producerConfigAccessor)
-        {
-            producer = new ProducerBuilder<string, string>(producerConfigAccessor.Value).Build();
-            TopicName = producerConfigAccessor.Value.TopicName;
-        }
-
-        public void Dispose()
-        {
-            producer.Dispose();
-        }
-
-        public async Task ProduceAsync(string notice,string terminalNo)
-        {
-            await producer.ProduceAsync(TopicName, new Message<string, string>
-            {
-                Key = notice,
-                Value = terminalNo
-            });
-        }
-    }
-}

src/JT808.Gateway.SimpleClient/JT808.Gateway.SimpleClient.csproj

@@ -1,18 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Grpc.Net.ClientFactory" Version="2.24.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.SimpleClient/Program.cs

@@ -1,52 +0,0 @@
-using JT808.Gateway.Client;
-using JT808.Gateway.SimpleClient.Services;
-using JT808.Protocol;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Threading.Tasks;
-using Grpc.Net.Client;
-using JT808.Gateway.GrpcService;
-using System.Net;
-
-namespace JT808.Gateway.SimpleClient
-{
-    class Program
-    {
-        static async Task Main(string[] args)
-        {
-            //ref https://docs.microsoft.com/zh-cn/aspnet/core/grpc/troubleshoot?view=aspnetcore-3.0#call-insecure-grpc-services-with-net-core-client
-            //ref https://docs.microsoft.com/zh-cn/aspnet/core/grpc/troubleshoot?view=aspnetcore-3.0
-
-            //先执行 dotnet dev-certs https --trust  命令生成开发证书
-            //使用 certmgr.msc 导出证书在服务端配置对应证书文件
-            //Uri "https://localhost:5001"
-
-            var serverHostBuilder = new HostBuilder()
-              .ConfigureLogging((context, logging) =>
-              {
-                  logging.AddConsole();
-                  logging.SetMinimumLevel(LogLevel.Trace);
-              })
-              .ConfigureServices((hostContext, services) =>
-              {
-                  services.AddGrpcClient<JT808Gateway.JT808GatewayClient>(o =>
-                  {
-                        o.Address = new Uri("https://localhost:5001");
-                  });
-                  services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                  services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                  services.AddLogging(options => {
-                      options.AddConsole();
-                      options.SetMinimumLevel(LogLevel.Trace);
-                  });
-                  services.AddJT808Configure()
-                          .AddJT808Client();
-                  services.AddHostedService<UpService>();
-                  //services.AddHostedService<GrpcClientService>();
-              });
-            await serverHostBuilder.RunConsoleAsync();
-        }
-    }
-}

src/JT808.Gateway.SimpleClient/Services/GrpcClientService.cs

@@ -1,68 +0,0 @@
-using JT808.Gateway.Client;
-using JT808.Protocol.MessageBody;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.DependencyInjection;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using JT808.Gateway.GrpcService;
-using static JT808.Gateway.GrpcService.JT808Gateway;
-using Google.Protobuf;
-using System.Text.Json;
-using JT808.Protocol.Extensions;
-
-namespace JT808.Gateway.SimpleClient.Services
-{
-    public class GrpcClientService : IHostedService
-    {
-        private readonly ILogger logger;
-        private readonly JT808GatewayClient client;
-
-        public GrpcClientService(
-            ILoggerFactory loggerFactory,
-            JT808GatewayClient  jT808GatewayClient)
-        {
-            this.client = jT808GatewayClient;
-            logger = loggerFactory.CreateLogger("GrpcClientService");
-        }
-
-        public Task StartAsync(CancellationToken cancellationToken)
-        {
-            Task.Run(() => {
-                //while (!cancellationToken.IsCancellationRequested)
-                //{
-                    Thread.Sleep(1000 * 10);
-                    var result1 = client.GetTcpAtomicCounter(new Empty());
-                    var result2 = client.GetUdpAtomicCounter(new Empty());
-                    var result3 = client.GetTcpSessionAll(new Empty());
-                    var result4 = client.GetUdpSessionAll(new Empty());
-                    var result5 = client.UnificationSend(new UnificationSendRequest() 
-                    {
-                         TerminalPhoneNo= "12345678910",
-                         Data= ByteString.CopyFrom("7E 02 00 00 26 12 34 56 78 90 12 00 7D 02 00 00 00 01 00 00 00 02 00 BA 7F 0E 07 E4 F1 1C 00 28 00 3C 00 00 18 10 15 10 10 10 01 04 00 00 00 64 02 02 00 7D 01 13 7E".ToHexBytes())
-                    });
-                    var result6 = client.RemoveSessionByTerminalPhoneNo(new  SessionRemoveRequest()
-                    {
-                         TerminalPhoneNo= "12345678910"
-                    });
-
-                    logger.LogDebug(JsonSerializer.Serialize(result1));
-                    logger.LogDebug(JsonSerializer.Serialize(result2));
-                    logger.LogDebug(JsonSerializer.Serialize(result3));
-                    logger.LogDebug(JsonSerializer.Serialize(result4));
-                    logger.LogDebug(JsonSerializer.Serialize(result5));
-                    logger.LogDebug(JsonSerializer.Serialize(result6));
-                //}  
-            }, cancellationToken);  
-            return Task.CompletedTask;
-        }
-
-        public Task StopAsync(CancellationToken cancellationToken)
-        {
-            return Task.CompletedTask;
-        }
-    }
-}

src/JT808.Gateway.SimpleClient/Services/UpService.cs

@@ -1,68 +0,0 @@
-using JT808.Gateway.Client;
-using JT808.Protocol.MessageBody;
-using Microsoft.Extensions.Hosting;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.SimpleClient.Services
-{
-    public class UpService : IHostedService
-    {
-        private readonly IJT808TcpClientFactory jT808TcpClientFactory;
-
-        public UpService(IJT808TcpClientFactory jT808TcpClientFactory)
-        {
-            this.jT808TcpClientFactory = jT808TcpClientFactory;
-        }
-
-        public Task StartAsync(CancellationToken cancellationToken)
-        {
-            JT808TcpClient client1 = jT808TcpClientFactory.Create(new DeviceConfig("12345678910", "127.0.0.1", 808));
-            //1.终端注册
-            client1.Send(new JT808_0x0100()
-            {
-                PlateNo = "粤A12345",
-                PlateColor = 2,
-                AreaID = 0,
-                CityOrCountyId = 0,
-                MakerId = "Koike001",
-                TerminalId = "Koike001",
-                TerminalModel = "Koike001"
-            });
-            //2.终端鉴权
-            client1.Send(new JT808_0x0102()
-            {
-                Code = "1234"
-            });
-            Task.Run(() => {
-                while (true)
-                {
-                    var i = 0;
-                    //3.每5000秒发一次
-                    client1.Send(new JT808_0x0200()
-                    {
-                        Lat = 110000 + i,
-                        Lng = 100000 + i,
-                        GPSTime = DateTime.Now,
-                        Speed = 50,
-                        Direction = 30,
-                        AlarmFlag = 5,
-                        Altitude = 50,
-                        StatusFlag = 10
-                    });
-                    i++;
-                    Thread.Sleep(5000);
-                }
-            });
-            return Task.CompletedTask;
-        }
-
-        public Task StopAsync(CancellationToken cancellationToken)
-        {
-            return Task.CompletedTask;
-        }
-    }
-}

src/JT808.Gateway.SimpleServer/Configs/nlog.Unix.config

@@ -1,36 +0,0 @@
-<?xml version="1.0" encoding="utf-8" ?>
-<!--
-      参考：http://www.cnblogs.com/fuchongjundream/p/3936431.html
-      autoReload：自动再配置
-      internalLogFile：可以让NLog把内部的调试和异常信息都写入指定文件里程序没问题了，日志却出了问题。这个该怎么办，到底是哪里不正确了？假如日志本身除了bug该如何解决？这就需要日志排错。把日志的错误信息写入日志。
-      <nlog throwExceptions="true" />    
-      <nlog internalLogFile="file.txt" />- 设置internalLogFile属性可以让NLog把内部的调试和异常信息都写入指定文件里。
-      <nlog internalLogLevel="Trace|Debug|Info|Warn|Error|Fatal" /> - 决定内部日志的级别，级别越高，输出的日志信息越简洁。
-      <nlog internalLogToConsole="false|true" /> - 是否把内部日志输出到标准控制台。
-      <nlog internalLogToConsoleError="false|true" /> - 是否把内部日志输出到标准错误控制台 (stderr)。
-      设置throwExceptions属性为“true”可以让NLog不再阻挡这类异常，而是把它们抛给调用者。在部署是这样做可以帮我们快速定位问题。一旦应用程序已经正确配置了，我们建议把throwExceptions的值设为“false”，这样由于日志引发的问题不至于导致应用程序的崩溃。
-  -->
-<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xsi:schemaLocation="NLog NLog.xsd"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    autoReload="true"
-    internalLogFile="/data/gps_services/serviceslogs/JT808.Gateway.SimpleServer/internalLog.txt"
-    internalLogLevel="Debug" >
-  <variable name="Directory" value="/data/gps_services/serviceslogs/JT808.Gateway.SimpleServer"/>
-  <targets>
-    <target name="SimpleServer" xsi:type="File"
-       fileName="${Directory}/SimpleServer.${shortdate}.log"
-       layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level}:${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}"/>
-    <target name="console" xsi:type="ColoredConsole"
-           useDefaultRowHighlightingRules="false"
-           layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level} ${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}">
-      <highlight-row condition="level == LogLevel.Debug" foregroundColor="DarkGray" />
-      <highlight-row condition="level == LogLevel.Info" foregroundColor="Gray" />
-      <highlight-row condition="level == LogLevel.Warn" foregroundColor="Yellow" />
-      <highlight-row condition="level == LogLevel.Error" foregroundColor="Red" />
-      <highlight-row condition="level == LogLevel.Fatal" foregroundColor="Red" backgroundColor="White" />
-    </target>
-  </targets>
-  <rules>
-    <logger name="*" minlevel="Debug" maxlevel="Fatal" writeTo="SimpleServer"/>
-  </rules>
-</nlog>

src/JT808.Gateway.SimpleServer/JT808.Gateway.SimpleServer.csproj

@@ -1,39 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Grpc.AspNetCore" Version="2.24.0" />
-    <PackageReference Include="NLog.Extensions.Logging" Version="1.6.1" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <None Update="appsettings.Development.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-    <None Update="appsettings.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-    <None Update="Configs\nlog.Unix.config">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-    <None Update="Configs\nlog.Win32NT.config">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-    <None Update="Configs\NLog.xsd">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-    <None Update="Configs\test.cer">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-  </ItemGroup>
-
-</Project>
-

src/JT808.Gateway.SimpleServer/Program.cs

@@ -1,71 +0,0 @@
-using JT808.Gateway.Services;
-using JT808.Gateway.Tcp;
-using JT808.Gateway.Udp;
-using JT808.Protocol;
-using Microsoft.AspNetCore.Builder;
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.AspNetCore.Server.Kestrel.Core;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using NLog.Extensions.Logging;
-using System;
-using System.IO;
-using System.Net;
-
-namespace JT808.Gateway.SimpleServer
-{
-    class Program
-    {
-        static void Main(string[] args)
-        {
-            Host.CreateDefaultBuilder(args)
-                .ConfigureAppConfiguration((hostingContext, config) =>
-                {
-                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                    config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true);
-                })
-                .ConfigureLogging((hostContext, configLogging) => {
-                    Console.WriteLine($"Environment.OSVersion.Platform:{Environment.OSVersion.Platform.ToString()}");
-                    NLog.LogManager.LoadConfiguration($"Configs/nlog.{Environment.OSVersion.Platform.ToString()}.config");
-                    configLogging.AddNLog(new NLogProviderOptions { CaptureMessageTemplates = true, CaptureMessageProperties = true });
-                    configLogging.SetMinimumLevel(LogLevel.Trace);
-                })
-                .ConfigureWebHostDefaults(webBuilder =>
-                {
-                    webBuilder
-                    .ConfigureKestrel(options =>
-                    {
-                        options.Listen(IPAddress.Any, 5001, listenOptions =>
-                        {
-                            listenOptions.Protocols = HttpProtocols.Http2;
-                            listenOptions.UseHttps($"{Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Configs", "test.cer")}", "");
-                        });
-                    })
-                    .Configure(app =>
-                    {
-                        app.UseRouting();
-                        app.UseEndpoints(endpoints =>
-                        {
-                            endpoints.MapGrpcService<JT808GatewayService>();
-                        });
-                    });
-                })
-                .ConfigureServices((hostContext,services) =>
-                {
-                    //services.Configure<KestrelServerOptions>(hostContext.Configuration.GetSection("Kestrel"));
-                    services.AddGrpc();
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                    services.AddJT808Configure()
-                            .AddJT808Gateway(hostContext.Configuration)
-                            .AddJT808GatewayTcpHost()
-                            .AddJT808GatewayUdpHost()
-                            .Builder();
-                })
-                .Build()
-                .Run();
-        }
-    }
-}

src/JT808.Gateway.Test/JT808.Gateway.Test.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="3.1.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
+    <PackageReference Include="System.IO.Pipelines" Version="4.7.0" />
+    <PackageReference Include="xunit" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="coverlet.collector" Version="1.1.0">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\JT808.Gateway\JT808.Gateway.csproj" />
+  </ItemGroup>
+
+</Project>

src/JT808.Gateway.Test/PipeTest.cs

@@ -0,0 +1,100 @@
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace JT808.Gateway.Test
+{
+    public class PipeTest
+    {
+        [Fact]
+        public void Test1()
+        {
+            var reader = new ReadOnlySequence<byte>(new byte[] { 0x7E, 0, 1, 2, 0x7E});
+            SequenceReader<byte> seqReader = new SequenceReader<byte>(reader);
+            int index = 0;
+            byte mark = 0;
+            long totalConsumed = 0;
+            List<byte[]> packages = new List<byte[]>();
+            while (!seqReader.End)
+            {
+                if (seqReader.IsNext(0x7E, advancePast: true))
+                {
+                    if (mark == 1)
+                    {
+                        var package = seqReader.Sequence.Slice(totalConsumed, seqReader.Consumed - totalConsumed).ToArray();
+                        packages.Add(package);
+                        totalConsumed += (seqReader.Consumed - totalConsumed);
+                        index++;
+                        if (seqReader.End) break;
+                        seqReader.Advance(1);
+                        mark = 0;
+                    }
+                    mark++;
+                }
+                else
+                {
+                    seqReader.Advance(1);
+                }
+                index++;
+            }
+            Assert.Equal(5, index);
+            Assert.Single(packages);
+            Assert.Equal(5, seqReader.Consumed);
+        }
+
+        [Fact]
+        public void Test2()
+        {
+            var reader = new ReadOnlySequence<byte>(new byte[] { 0x7E, 0, 1, 2, 0x7E, 0x7E, 0, 1, 0x7E, 0x7E, 2, 2, 2 });
+            SequenceReader<byte> seqReader = new SequenceReader<byte>(reader);
+            int index = 0;
+            byte mark = 0;
+            long totalConsumed = 0;
+            List<byte[]> packages = new List<byte[]>();
+            while (!seqReader.End)
+            {
+                if (seqReader.IsNext(0x7E, advancePast: true))
+                {
+                    if (mark == 1)
+                    {
+                        var package = seqReader.Sequence.Slice(totalConsumed, seqReader.Consumed - totalConsumed).ToArray();
+                        packages.Add(package);
+                        totalConsumed += (seqReader.Consumed - totalConsumed);
+                        index++;
+                        if (seqReader.End) break;
+                        seqReader.Advance(1);
+                        mark = 0;
+                    }
+                    mark++;
+                }
+                else
+                {
+                    seqReader.Advance(1);
+                }
+                index++;
+            }
+            Assert.Equal(13, index);
+            Assert.Equal(2,packages.Count);
+            Assert.Equal(9, totalConsumed);
+            Assert.Equal(13, seqReader.Consumed);
+        }
+
+        [Fact]
+        public void Test3()
+        {
+            Assert.Throws<Exception>(() => 
+            {
+                var reader = new ReadOnlySequence<byte>(new byte[] { 0, 1, 2, 0x7E });
+                SequenceReader<byte> seqReader = new SequenceReader<byte>(reader);
+                if (seqReader.TryPeek(out byte beginMark))
+                {
+                    if (beginMark != 0x7E) throw new ArgumentException("not 808 packages");
+                }
+            });
+        }
+    }
+}

src/JT808.Gateway.Test/Session/JT808SessionManagerTest.cs

@@ -0,0 +1,187 @@
+using JT808.Gateway.Session;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Xunit;
+using Microsoft.Extensions.Logging;
+using System.Net.Sockets;
+
+namespace JT808.Gateway.Test.Session
+{
+    public class JT808SessionManagerTest
+    {
+        [Fact]
+        public void TryAddTest()
+        {
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var result=jT808SessionManager.TryAdd(new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp)));
+            Assert.True(result);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+        }
+
+        [Fact]
+        public void TryLinkTest()
+        {
+            string tno = "123456";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session);
+            jT808SessionManager.TryLink(tno, session);
+            Assert.True(result1);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+            Assert.True(jT808SessionManager.TerminalPhoneNoSessions.ContainsKey(tno));
+        }
+
+        /// <summary>
+        /// 用于转发过来的车辆
+        /// </summary>
+        [Fact]
+        public void TryLinkTest1_1_N()
+        {
+            string tno1 = "123456";
+            string tno2 = "123457";
+            string tno3 = "123458";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session);
+            jT808SessionManager.TryLink(tno1, session);
+            jT808SessionManager.TryLink(tno2, session);
+            jT808SessionManager.TryLink(tno3, session);
+            Assert.True(result1);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+            Assert.Equal(3,jT808SessionManager.TerminalPhoneNoSessions.Count);
+            jT808SessionManager.RemoveBySessionId(session.SessionID);
+            Assert.Equal(0, jT808SessionManager.TotalSessionCount);
+            Assert.Empty(jT808SessionManager.TerminalPhoneNoSessions);
+        }
+
+        /// <summary>
+        /// 用于转发过来的车辆
+        /// </summary>
+        [Fact]
+        public void TryLinkTest2_1_N()
+        {
+            string tno1 = "123456";
+            string tno2 = "123457";
+            string tno3 = "123458";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session);
+            jT808SessionManager.TryLink(tno1, session);
+            jT808SessionManager.TryLink(tno2, session);
+            jT808SessionManager.TryLink(tno3, session);
+            Assert.True(result1);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+            Assert.Equal(3, jT808SessionManager.TerminalPhoneNoSessions.Count);
+            jT808SessionManager.RemoveByTerminalPhoneNo(tno1);
+            Assert.Equal(0, jT808SessionManager.TotalSessionCount);
+            Assert.Empty(jT808SessionManager.TerminalPhoneNoSessions);
+        }
+
+        /// <summary>
+        /// 转发过来的车辆切换为直连车辆
+        /// </summary>
+        [Fact]
+        public void UpdateLinkTest2_1_N()
+        {
+            string tno1 = "123456";
+            string tno2 = "123457";
+            string tno3 = "123458";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session1 = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var session2 = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session1);
+            var result2 = jT808SessionManager.TryAdd(session2);
+            //转发车辆
+            jT808SessionManager.TryLink(tno1, session1);
+            jT808SessionManager.TryLink(tno2, session1);
+            //直连车辆
+            jT808SessionManager.TryLink(tno3, session2);
+
+            Assert.True(result1);
+            Assert.True(result2);
+            Assert.Equal(2, jT808SessionManager.TotalSessionCount);
+            Assert.Equal(3, jT808SessionManager.TerminalPhoneNoSessions.Count);
+
+            //将tno2切换为直连车辆
+            var session3 = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result3 = jT808SessionManager.TryAdd(session3);
+            jT808SessionManager.TryLink(tno2, session3);
+            Assert.True(result3);
+            if (jT808SessionManager.TerminalPhoneNoSessions.TryGetValue(tno2,out string sessionid))
+            {
+                //实际的通道Id
+                Assert.Equal(session3.SessionID, sessionid);
+            }
+            Assert.Equal(3, jT808SessionManager.TotalSessionCount);
+            Assert.Equal(3, jT808SessionManager.TerminalPhoneNoSessions.Count);
+
+            jT808SessionManager.RemoveByTerminalPhoneNo(tno1);
+            Assert.Equal(2, jT808SessionManager.TotalSessionCount);
+            Assert.Equal(2,jT808SessionManager.TerminalPhoneNoSessions.Count);
+        }
+
+        [Fact]
+        public void RemoveBySessionIdTest()
+        {
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session);
+            Assert.True(result1);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+            jT808SessionManager.RemoveBySessionId(session.SessionID);
+            Assert.Equal(0, jT808SessionManager.TotalSessionCount);
+        }
+
+        [Fact]
+        public void RemoveByTerminalPhoneNoTest()
+        {
+            string tno = "123456";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session);
+            jT808SessionManager.TryLink(tno, session);
+            Assert.True(result1);
+            Assert.Equal(1, jT808SessionManager.TotalSessionCount);
+            jT808SessionManager.RemoveByTerminalPhoneNo(tno);
+            Assert.False(jT808SessionManager.TerminalPhoneNoSessions.ContainsKey(tno));
+            Assert.Equal(0, jT808SessionManager.TotalSessionCount);
+        }
+
+        [Fact]
+        public void SendTest()
+        {
+            Assert.Throws<SocketException>(() => 
+            {
+                string tno = "123456";
+                JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+                var session = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+                var result1 = jT808SessionManager.TryAdd(session);
+                jT808SessionManager.TryLink(tno, session);
+                jT808SessionManager.TrySendByTerminalPhoneNo(tno, new byte[] { 0x7e, 0, 0, 0x7e });
+            });
+        }
+
+        [Fact]
+        public void GetTcpAllTest()
+        {
+            string tno1 = "123456";
+            string tno2 = "123457";
+            JT808SessionManager jT808SessionManager = new JT808SessionManager(new LoggerFactory());
+            var session1 = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var session2 = new JT808TcpSession(new Socket(SocketType.Stream, ProtocolType.Tcp));
+            var result1 = jT808SessionManager.TryAdd(session1);
+            var result2 = jT808SessionManager.TryAdd(session2);
+            jT808SessionManager.TryLink(tno1, session1);
+            jT808SessionManager.TryLink(tno2, session2);
+            Assert.True(result1);
+            Assert.True(result2);
+            Assert.Equal(2, jT808SessionManager.TotalSessionCount);
+            Assert.True(jT808SessionManager.TerminalPhoneNoSessions.ContainsKey(tno1));
+            Assert.True(jT808SessionManager.TerminalPhoneNoSessions.ContainsKey(tno2));
+            var sessions = jT808SessionManager.GetTcpAll();
+            Assert.Equal(session1.SessionID, sessions[0].SessionID);
+            Assert.Equal(session2.SessionID, sessions[1].SessionID);
+        }
+    }
+}

src/JT808.Gateway.CleintBenchmark/Configs/nlog.unix.config → src/JT808.Gateway.TestHosting/Configs/nlog.Unix.config

@@ -13,13 +13,13 @@
   <nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xsi:schemaLocation="NLog NLog.xsd"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       autoReload="true"
-      internalLogFile="/data/logs/JT808.DotNetty.Hosting/internalLog.txt"
+      internalLogFile="/data/serviceslogs/JT808.Gateway/internalLog.txt"
       internalLogLevel="Debug" >
-    <variable name="Directory" value="/data/logs/JT808.DotNetty.Hosting"/>
+    <variable name="Directory" value="/data/serviceslogs/JT808.Gateway"/>
     <targets>
-      <target name="all" xsi:type="File"
-             fileName="${Directory}/all/${shortdate}.log"
-             layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level}:${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}"/>
+           <target name="Gateway" xsi:type="File"
+              fileName="${Directory}/Gateway.${shortdate}.log"
+              layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level}:${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}"/>
       <target name="console" xsi:type="ColoredConsole"
              useDefaultRowHighlightingRules="false"
              layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level} ${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}">
@@ -31,6 +31,6 @@
       </target>
     </targets>
     <rules>
-      <logger name="*" minlevel="Debug" maxlevel="Fatal" writeTo="all,console"/>
+      <logger name="*" minlevel="Debug" maxlevel="Fatal" writeTo="Gateway"/>
     </rules>
   </nlog>

src/JT808.Gateway.SimpleServer/Configs/nlog.Win32NT.config → src/JT808.Gateway.TestHosting/Configs/nlog.Win32NT.config

@@ -13,12 +13,12 @@
   <nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xsi:schemaLocation="NLog NLog.xsd"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       autoReload="true"
-      internalLogFile="wwwroot/logs/JT808.Gateway.SimpleServer/internalLog.txt"
+      internalLogFile="wwwroot/logs/JT808.Gateway/internalLog.txt"
       internalLogLevel="Debug" >
-    <variable name="Directory" value="/wwwroot/logs/JT808.Gateway.SimpleServer"/>
+    <variable name="Directory" value="/data/logs/JT808.Gateway"/>
     <targets>
-      <target name="SimpleServer" xsi:type="File"
-              fileName="${Directory}/SimpleServer.${shortdate}.log"
+      <target name="Gateway" xsi:type="File"
+              fileName="${Directory}/Gateway.${shortdate}.log"
               layout="${date:format=yyyyMMddHHmmss} ${callsite} ${level}:${message} ${onexception:${exception:format=tostring} ${newline} ${stacktrace} ${newline}"/>
       <target name="console" xsi:type="ColoredConsole"
              useDefaultRowHighlightingRules="false"
@@ -31,6 +31,6 @@
       </target>
     </targets>
     <rules>
-      <logger name="*" minlevel="Trace" maxlevel="Fatal" writeTo="SimpleServer,console"/>
+      <logger name="*" minlevel="Debug" maxlevel="Fatal" writeTo="Gateway,console"/>
     </rules>
   </nlog>

src/JT808.Gateway.TestHosting/Dockerfile

@@ -0,0 +1,23 @@
+#See https://aka.ms/containerfastmode to understand how Visual Studio uses this Dockerfile to build your images for faster debugging.
+
+FROM mcr.microsoft.com/dotnet/core/runtime:3.1-buster-slim AS base
+EXPOSE 808/tcp
+WORKDIR /app
+
+FROM mcr.microsoft.com/dotnet/core/sdk:3.1-buster AS build
+WORKDIR /src
+COPY ["JT808.Gateway.TestHosting/JT808.Gateway.TestHosting.csproj", "JT808.Gateway.TestHosting/"]
+COPY ["JT808.Gateway/JT808.Gateway.csproj", "JT808.Gateway/"]
+COPY ["JT808.Gateway.Abstractions/JT808.Gateway.Abstractions.csproj", "JT808.Gateway.Abstractions/"]
+RUN dotnet restore "JT808.Gateway.TestHosting/JT808.Gateway.TestHosting.csproj"
+COPY . .
+WORKDIR "/src/JT808.Gateway.TestHosting"
+RUN dotnet build "JT808.Gateway.TestHosting.csproj" -c Release -o /app/build
+
+FROM build AS publish
+RUN dotnet publish "JT808.Gateway.TestHosting.csproj" -c Release -o /app/publish
+
+FROM base AS final
+WORKDIR /app
+COPY --from=publish /app/publish .
+ENTRYPOINT ["dotnet", "JT808.Gateway.TestHosting.dll"]

src/JT808.Gateway.CleintBenchmark/JT808.Gateway.CleintBenchmark.csproj → src/JT808.Gateway.TestHosting/JT808.Gateway.TestHosting.csproj

@@ -1,16 +1,18 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+    <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
   </PropertyGroup>
+
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Configuration.FileExtensions" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-    <PackageReference Include="NLog.Extensions.Logging" Version="1.6.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.1.0" />
+    <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.9.10" />
+    <PackageReference Include="NLog.Extensions.Logging" Version="1.6.1" />
   </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\JT808.Gateway\JT808.Gateway.csproj" />
   </ItemGroup>
@@ -19,10 +21,10 @@
     <None Update="appsettings.json">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
-    <None Update="Configs\nlog.unix.config">
+    <None Update="Configs\nlog.Unix.config">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
-    <None Update="Configs\nlog.win.config">
+    <None Update="Configs\nlog.Win32NT.config">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
     <None Update="Configs\NLog.xsd">

src/JT808.Gateway.TestHosting/Jobs/CallGrpcClientJob.cs

@@ -0,0 +1,69 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using JT808.Gateway.Configurations;
+using JT808.Gateway.GrpcService;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using System.Text.Json;
+
+namespace JT808.Gateway.TestHosting.Jobs
+{
+    public class CallGrpcClientJob :IHostedService
+    {
+        private Channel channel;
+        private readonly ILogger Logger;
+        private Grpc.Core.Metadata AuthMetadata;
+        public CallGrpcClientJob(
+            ILoggerFactory loggerFactory,
+            JT808Configuration configuration)
+        {
+            Logger = loggerFactory.CreateLogger("CallGrpcClientJob");
+            channel = new Channel($"{configuration.WebApiHost}:{configuration.WebApiPort}",
+                ChannelCredentials.Insecure);
+            AuthMetadata = new Grpc.Core.Metadata();
+            AuthMetadata.Add("token", configuration.WebApiToken);
+        }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            Task.Run(() =>
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    JT808Gateway.JT808GatewayClient jT808GatewayClient = new JT808Gateway.JT808GatewayClient(channel);
+                    try
+                    {
+                        var result1 = jT808GatewayClient.GetTcpAtomicCounter(new Empty(), AuthMetadata);
+                        var result2 = jT808GatewayClient.GetTcpSessionAll(new Empty(), AuthMetadata);
+                        Logger.LogInformation($"[GetTcpAtomicCounter]:{JsonSerializer.Serialize(result1)}");
+                        Logger.LogInformation($"[GetTcpSessionAll]:{JsonSerializer.Serialize(result2)}");
+                    }
+                    catch (Exception ex)
+                    {
+                        Logger.LogError(ex, "Call Grpc Error");
+                    }
+                    try
+                    {
+                        var result1 = jT808GatewayClient.GetTcpAtomicCounter(new Empty());
+                    }
+                    catch (RpcException ex)
+                    {
+                        Logger.LogError($"{ex.StatusCode.ToString()}-{ex.Message}");
+                    }
+                    Thread.Sleep(3000);
+                }
+            }, cancellationToken);
+            return Task.CompletedTask;
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            channel.ShutdownAsync();
+            return Task.CompletedTask;
+        }
+    }
+}

src/JT808.Gateway.TestHosting/Program.cs

@@ -0,0 +1,46 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using JT808.Protocol;
+using Microsoft.Extensions.Configuration;
+using NLog.Extensions.Logging;
+using JT808.Gateway.TestHosting.Jobs;
+
+namespace JT808.Gateway.TestHosting
+{
+    class Program
+    {
+        static async Task Main(string[] args)
+        {
+            var serverHostBuilder = new HostBuilder()
+                .ConfigureAppConfiguration((hostingContext, config) =>
+                {
+                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory)
+                          .AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
+                          .AddJsonFile($"appsettings.{ hostingContext.HostingEnvironment.EnvironmentName}.json", optional: true, reloadOnChange: true);
+                })
+                .ConfigureLogging((context, logging) =>
+                {
+                    Console.WriteLine($"Environment.OSVersion.Platform:{Environment.OSVersion.Platform.ToString()}");
+                    NLog.LogManager.LoadConfiguration($"Configs/nlog.{Environment.OSVersion.Platform.ToString()}.config");
+                    logging.AddNLog(new NLogProviderOptions { CaptureMessageTemplates = true, CaptureMessageProperties = true });
+                    logging.SetMinimumLevel(LogLevel.Trace);
+                })
+                .ConfigureServices((hostContext, services) =>
+                {
+                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
+                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
+                    services.AddJT808Configure()
+                            .AddJT808Gateway()
+                            .AddTcp()
+                            .AddUdp()
+                            .AddGrpc();
+                    //services.AddHostedService<CallGrpcClientJob>();
+                });
+
+            await serverHostBuilder.RunConsoleAsync();
+        }
+    }
+}

src/JT808.Gateway.TestHosting/startup.txt

@@ -0,0 +1,4 @@
+pm2 start "dotnet JT808.DotNetty.CleintBenchmark.dll ASPNETCORE_ENVIRONMENT=Production" --max-restarts=1 -n "JT808.DotNetty.CleintBenchmark" -o "/data/pm2Logs/JT808.DotNetty.CleintBenchmark/out.log" -e "/data/pm2Logs/JT808.DotNetty.CleintBenchmark/error.log"
+
+
+pm2 start "dotnet JT808.Gateway.TestHosting.dll ASPNETCORE_ENVIRONMENT=Production" --max-restarts=1 -n "JT808.Gateway.808" -o "/data/pm2Logs/JT808.Gateway/out.log" -e "/data/pm2Logs/JT808.Gateway/error.log"

src/JT808.Gateway.Tests/JT808.Gateway.MsgIdHandler.Test/JT808.Gateway.MsgIdHandler.Test.csproj

@@ -1,24 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.0.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\..\JT808.Gateway.Kafka\JT808.Gateway.Kafka.csproj" />
-    <ProjectReference Include="..\..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-  <ItemGroup>
-    <None Update="appsettings.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.Tests/JT808.Gateway.MsgIdHandler.Test/JT808DotNettyMsgIdHandlerDefaultImpl.cs

@@ -1,44 +0,0 @@
-using JT808.Gateway.BusinessServices.MsgIdHandler;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.Kafka;
-using JT808.Gateway.PubSub;
-using JT808.Protocol;
-using JT808.Protocol.Extensions;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.MsgIdHandler.Test
-{
-    public class JT808DotNettyMsgIdHandlerDefaultImpl : IJT808MsgIdHandler
-    {
-        public readonly ILogger<JT808DotNettyMsgIdHandlerDefaultImpl> logger;
-        public JT808DotNettyMsgIdHandlerDefaultImpl(ILoggerFactory loggerFactory,
-                                                                        IServiceProvider serviceProvider) {
-            logger = loggerFactory.CreateLogger<JT808DotNettyMsgIdHandlerDefaultImpl>();
-            Task.Run(()=> {
-                while (true)
-                {
-                    Thread.Sleep(5000);
-                    using (IJT808MsgProducer jT808MsgProducer = new JT808MsgProducer(new JT808MsgProducerConfig
-                    {
-                        BootstrapServers = "127.0.0.1:9092",
-                        TopicName = "JT808Msg"
-                    }))
-                    {
-                        jT808MsgProducer.ProduceAsync("123456", new byte[] { 0x7E, 0, 0x7E }).Wait();
-                    }
-                }
-            });
-        }
-
-        public void Processor((string TerminalNo, byte[] Data) parameter)
-        {
-            logger.LogDebug($"{parameter.TerminalNo}:{parameter.Data.ToHexString()}");
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.MsgIdHandler.Test/Program.cs

@@ -1,39 +0,0 @@
-using JT808.Gateway.BusinessServices.MsgIdHandler;
-using JT808.Gateway.Kafka;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.MsgIdHandler.Test
-{
-    class Program
-    {
-       async static Task Main(string[] args)
-        {
-            var serverHostBuilder = new HostBuilder()
-                .UseEnvironment(args[0].Split('=')[1])
-                .ConfigureAppConfiguration((hostingContext,config) => {
-                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                    config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
-                             .AddJsonFile($"appsettings.{ hostingContext.HostingEnvironment.EnvironmentName}.json", optional: true, reloadOnChange: true);
-                })
-                .ConfigureLogging(configLogging => {
-                    configLogging.AddConsole();
-                    configLogging.SetMinimumLevel(LogLevel.Trace);
-                })
-                .ConfigureServices((hostContext, services) => {
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                    services.AddJT808Configure()
-                                .AddJT808ClientKafka()
-                                .AddMsgConsumer(hostContext.Configuration)
-                                .AddJT808MsgIdHandler<JT808DotNettyMsgIdHandlerDefaultImpl>();
-                });
-            await serverHostBuilder.RunConsoleAsync();
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.MsgLogging.Test/JT808.Gateway.MsgLogging.Test.csproj

@@ -1,25 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.0.0" />
-  </ItemGroup>
-  
-  <ItemGroup>
-    <ProjectReference Include="..\..\JT808.Gateway.Kafka\JT808.Gateway.Kafka.csproj" />
-    <ProjectReference Include="..\..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-  
-  <ItemGroup>
-    <None Update="appsettings.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.Tests/JT808.Gateway.MsgLogging.Test/JT808MsgLoggingImpl.cs

@@ -1,52 +0,0 @@
-using JJT808.Gateway.Kafka;
-using JT808.Gateway.BusinessServices.MsgLogging;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.Kafka;
-using JT808.Gateway.PubSub;
-using JT808.Protocol.Extensions;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.MsgLogging.Test
-{
-    public class JT808MsgLoggingImpl : IJT808MsgLogging
-    {
-        public readonly ILogger<JT808MsgLoggingImpl> logger;
-        public JT808MsgLoggingImpl(ILoggerFactory loggerFactory) {
-            logger = loggerFactory.CreateLogger<JT808MsgLoggingImpl>();
-            Task.Run(() => {
-                while (true)
-                {
-                    Thread.Sleep(5000);
-                    using (IJT808MsgProducer jT808MsgProducer = new JT808MsgProducer(new JT808MsgProducerConfig
-                    {
-                        BootstrapServers = "127.0.0.1:9092",
-                        TopicName = "JT808Msg"
-                    }))
-                    {
-                        jT808MsgProducer.ProduceAsync("123456", new byte[] { 0x7E, 0,0,0,0, 0x7E }).Wait();
-                    }
-
-                    JT808MsgReplyProducerConfig JT808MsgProducerConfig = new JT808MsgReplyProducerConfig
-                    {
-                        TopicName = "JT808MsgReply",
-                        BootstrapServers = "127.0.0.1:9092",
-                    };
-                    using (IJT808MsgReplyProducer jT808MsgProducer = new JT808MsgReplyProducer(JT808MsgProducerConfig))
-                    {
-                        jT808MsgProducer.ProduceAsync("123456", new byte[] { 0x7E,1,1,1,1, 0x7E }).Wait();
-                    }
-                }
-            });
-        }
-
-        public void Processor((string TerminalNo, byte[] Data) parameter, JT808MsgLoggingType jT808MsgLoggingType)
-        {
-            logger.LogDebug($"{parameter.TerminalNo}:{parameter.Data.ToHexString()},方向:{jT808MsgLoggingType.ToString()}");
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.MsgLogging.Test/Program.cs

@@ -1,43 +0,0 @@
-using JT808.Gateway.BusinessServices.MsgLogging;
-using JT808.Gateway.Kafka;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Logging.Console;
-using System;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.MsgLogging.Test
-{
-    class Program
-    {
-       async static Task Main(string[] args)
-        {
-            var hostBuilder = new HostBuilder()
-                .UseEnvironment(args[0].Split('=')[1])
-                .ConfigureAppConfiguration((hostContext,config)=> {
-                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                    config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
-                             .AddJsonFile($"appsettings.{hostContext.HostingEnvironment.EnvironmentName}.json", optional: true, reloadOnChange: true);
-                })
-                .ConfigureLogging((hostContext, configLogging) => {
-                    configLogging.AddConsole();
-                    configLogging.SetMinimumLevel(LogLevel.Trace);
-                })
-                .ConfigureServices((hostContext, services) => {
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                    services.AddJT808Configure()
-                                .AddJT808ClientKafka()
-                                .AddMsgConsumer(hostContext.Configuration)
-                                .AddMsgReplyConsumer(hostContext.Configuration)
-                                .AddJT808MsgLogging<JT808MsgLoggingImpl>();
-                })
-                ;
-
-            await hostBuilder.RunConsoleAsync();
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.ReplyMessage.Test/JT808.Gateway.ReplyMessage.Test.csproj

@@ -1,25 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.0.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\..\JT808.Gateway.Kafka\JT808.Gateway.Kafka.csproj" />
-    <ProjectReference Include="..\..\JT808.Gateway\JT808.Gateway.csproj" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <None Update="appsettings.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.Tests/JT808.Gateway.ReplyMessage.Test/JT808DotNettyReplyMessageServiceInherited.cs

@@ -1,48 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using JT808.Gateway.BusinessServices.ReplyMessage;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.Kafka;
-using JT808.Gateway.PubSub;
-using JT808.Protocol;
-using JT808.Protocol.Extensions;
-using Microsoft.Extensions.Logging;
-
-namespace JT808.Gateway.ReplyMessage.Test
-{
-    public class JT808DotNettyReplyMessageServiceInherited : JT808ReplyMessageService
-    {
-        public readonly ILogger<JT808DotNettyReplyMessageServiceInherited> logger;
-
-        public JT808DotNettyReplyMessageServiceInherited(IJT808Config jT808Config,
-                                                                                        IJT808MsgReplyProducer jT808MsgReplyProducer,
-                                                                                        ILoggerFactory loggerFactory) 
-            : base(jT808Config, jT808MsgReplyProducer)
-        {
-            logger = loggerFactory.CreateLogger<JT808DotNettyReplyMessageServiceInherited>();
-            Task.Run(() => {
-                while (true)
-                {
-                    Thread.Sleep(5000);
-                    using (IJT808MsgProducer jT808MsgProducer = new JT808MsgProducer(new JT808MsgProducerConfig
-                    {
-                        BootstrapServers = "127.0.0.1:9092",
-                        TopicName = "JT808Msg"
-                    }))
-                    {
-                        jT808MsgProducer.ProduceAsync("011111111111", "7E02000032011111111111012E00000000000C00000160E42506C30C82002C00000000180914142057010400001DC003020000250400000000300115310100977E".ToHexBytes()).Wait();
-                    }
-                }
-            });
-        }
-
-        public override void Processor((string TerminalNo, byte[] Data) parameter)
-        {
-            logger.LogDebug($"{parameter.TerminalNo}:{parameter.Data.ToHexString()}");
-            base.Processor(parameter);
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.ReplyMessage.Test/Program.cs

@@ -1,42 +0,0 @@
-using JT808.Gateway.BusinessServices.ReplyMessage;
-using JT808.Gateway.Kafka;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.ReplyMessage.Test
-{
-    class Program
-    {
-        async static Task Main(string[] args)
-        {
-            var hostBuilder = new HostBuilder()
-                .UseEnvironment(args[0].Split('=')[1])
-                .ConfigureAppConfiguration((hostContext, config) => {
-                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                    config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
-                             .AddJsonFile($"appsettings.{hostContext.HostingEnvironment.EnvironmentName}.json", optional: true, reloadOnChange: true);
-                })
-                .ConfigureLogging((hostContext, configLogging) => {
-                    configLogging.AddConsole();
-                    configLogging.SetMinimumLevel(LogLevel.Trace);
-                })
-                .ConfigureServices((hostContext, services) => {
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                    services.AddJT808Configure()
-                                .AddJT808ClientKafka()
-                                .AddMsgConsumer(hostContext.Configuration)
-                                .AddMsgReplyProducer(hostContext.Configuration)
-                                .AddInprocJT808ReplyMessage<JT808DotNettyReplyMessageServiceInherited>();
-                })
-                ;
-
-            await hostBuilder.RunConsoleAsync();
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.SessionNotice.Test/JT808.Gateway.SessionNotice.Test.csproj

@@ -1,24 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.0.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\..\JT808.Gateway.Kafka\JT808.Gateway.Kafka.csproj" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <None Update="appsettings.json">
-      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
-    </None>
-  </ItemGroup>
-
-</Project>

src/JT808.Gateway.Tests/JT808.Gateway.SessionNotice.Test/JT808DotNettySessionNoticeServiceInherited.cs

@@ -1,41 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using JT808.Gateway.BusinessServices.SessionNotice;
-using JT808.Gateway.Configs.Kafka;
-using JT808.Gateway.Kafka;
-using JT808.Gateway.PubSub;
-using Microsoft.Extensions.Logging;
-
-namespace JT808.Gateway.SessionNotice.Test
-{
-    public class JT808DotNettySessionNoticeServiceInherited : JT808SessionNoticeService
-    {
-        public JT808DotNettySessionNoticeServiceInherited(ILoggerFactory loggerFactory) : base(loggerFactory)
-        {
-            Task.Run(()=> {
-                while (true)
-                {
-                    Thread.Sleep(5000);
-                    JT808SessionProducerConfig JT808ProducerConfig = new JT808SessionProducerConfig
-                    {
-                        TopicName = "JT808Session",
-                        BootstrapServers = "127.0.0.1:9092"
-                    };
-                    using (IJT808SessionProducer jT808MsgProducer = new JT808SessionProducer(JT808ProducerConfig))
-                    {
-                        jT808MsgProducer.ProduceAsync("online", "123456").Wait();
-                        jT808MsgProducer.ProduceAsync("offline", "123457").Wait();
-                    }
-                }         
-            });
-        }
-
-        public override void Processor((string Notice, string TerminalNo) parameter)
-        {
-            base.Processor(parameter);
-        }
-    }
-}

src/JT808.Gateway.Tests/JT808.Gateway.SessionNotice.Test/Program.cs

@@ -1,41 +0,0 @@
-using JT808.Gateway.BusinessServices.SessionNotice;
-using JT808.Gateway.Kafka;
-using JT808.Protocol;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Threading.Tasks;
-
-namespace JT808.Gateway.SessionNotice.Test
-{
-    class Program
-    {
-        async static Task Main(string[] args)
-        {
-            var hostBuilder = new HostBuilder()
-                .UseEnvironment(args[0].Split('=')[1])
-                .ConfigureAppConfiguration((hostContext, config) => {
-                    config.SetBasePath(AppDomain.CurrentDomain.BaseDirectory);
-                    config.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
-                             .AddJsonFile($"appsettings.{hostContext.HostingEnvironment.EnvironmentName}.json", optional: true, reloadOnChange: true);
-                })
-                .ConfigureLogging((hostContext, configLogging) => {
-                    configLogging.AddConsole();
-                    configLogging.SetMinimumLevel(LogLevel.Trace);
-                })
-                .ConfigureServices((hostContext, services) => {
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
-                    services.AddJT808Configure()
-                                .AddJT808ClientKafka()
-                                .AddSessionConsumer(hostContext.Configuration)
-                                .AddInprocJT808SessionNotice<JT808DotNettySessionNoticeServiceInherited>();
-                })
-                ;
-
-            await hostBuilder.RunConsoleAsync();
-        }
-    }
-}