服务测试用例

src/JT808.DotNetty.Kafka/JT808ClientKafkaExtensions.cs

@@ -39,6 +39,18 @@ namespace JT808.DotNetty.Kafka
         /// <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>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.MsgIdHandler.Test/JT808.DotNetty.MsgIdHandler.Test.csproj

@@ -5,4 +5,20 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.MsgIdHandler\JT808.DotNetty.MsgIdHandler.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.MsgIdHandler.Test/JT808DotNettyMsgIdHandlerDefaultImpl.cs

@@ -0,0 +1,42 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+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.DotNetty.MsgIdHandler.Test
+{
+    public class JT808DotNettyMsgIdHandlerDefaultImpl : IJT808DotNettyMsgIdHandler
+    {
+        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.DotNetty.Services.Tests/JT808.DotNetty.MsgIdHandler.Test/Program.cs

@@ -1,12 +1,38 @@
-using System;
+using JT808.DotNetty.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.DotNetty.MsgIdHandler.Test
 {
     class Program
     {
-        static void Main(string[] args)
+       async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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.DotNetty.Services.Tests/JT808.DotNetty.MsgLogging.Test/JT808.DotNetty.MsgLogging.Test.csproj

@@ -5,4 +5,21 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.MsgLogging\JT808.DotNetty.MsgLogging.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.MsgLogging.Test/JT808MsgLoggingImpl.cs

@@ -0,0 +1,49 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+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.DotNetty.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.DotNetty.Services.Tests/JT808.DotNetty.MsgLogging.Test/Program.cs

@@ -1,12 +1,42 @@
-using System;
+using JT808.DotNetty.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.DotNetty.MsgLogging.Test
 {
     class Program
     {
-        static void Main(string[] args)
+       async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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.DotNetty.Services.Tests/JT808.DotNetty.ReplyMessage.Test/JT808.DotNetty.ReplyMessage.Test.csproj

@@ -5,4 +5,21 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.ReplyMessage\JT808.DotNetty.ReplyMessage.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.ReplyMessage.Test/JT808DotNettyReplyMessageServiceInherited.cs

@@ -0,0 +1,46 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+using JT808.Protocol;
+using JT808.Protocol.Extensions;
+using Microsoft.Extensions.Logging;
+
+namespace JT808.DotNetty.ReplyMessage.Test
+{
+    public class JT808DotNettyReplyMessageServiceInherited : JT808DotNettyReplyMessageService
+    {
+        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.DotNetty.Services.Tests/JT808.DotNetty.ReplyMessage.Test/Program.cs

@@ -1,12 +1,41 @@
-using System;
+using JT808.DotNetty.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.DotNetty.ReplyMessage.Test
 {
     class Program
     {
-        static void Main(string[] args)
+        async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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.DotNetty.Services.Tests/JT808.DotNetty.SessionNotice.Test/JT808.DotNetty.SessionNotice.Test.csproj

@@ -5,4 +5,21 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.SessionNotice\JT808.DotNetty.SessionNotice.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.SessionNotice.Test/JT808DotNettySessionNoticeServiceInherited.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+using Microsoft.Extensions.Logging;
+
+namespace JT808.DotNetty.SessionNotice.Test
+{
+    public class JT808DotNettySessionNoticeServiceInherited : JT808DotNettySessionNoticeService
+    {
+        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.DotNetty.Services.Tests/JT808.DotNetty.SessionNotice.Test/Program.cs

@@ -1,12 +1,40 @@
-using System;
+using JT808.DotNetty.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.DotNetty.SessionNotice.Test
 {
     class Program
     {
-        static void Main(string[] args)
+        async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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();
         }
     }
 }

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

@@ -5,4 +5,21 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.Traffic\JT808.DotNetty.Traffic.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.Traffic.Test/JT808DotNettyTrafficServiceTest.cs

@@ -0,0 +1,39 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+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.DotNetty.Traffic.Test
+{
+    public class JT808DotNettyTrafficServiceTest
+    {
+        private readonly CSRedis.CSRedisClient redisClien;
+        public readonly ILogger<JT808DotNettyTrafficServiceTest> logger;
+        public JT808DotNettyTrafficServiceTest(ILoggerFactory loggerFactory) {
+            redisClien = new CSRedis.CSRedisClient("127.0.0.1:6379,password=smallchi");
+            RedisHelper.Initialization(redisClien);
+            logger = loggerFactory.CreateLogger<JT808DotNettyTrafficServiceTest>();
+            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();
+                    }
+                   var length= RedisHelper.HGet("011111111111", DateTime.Now.ToString("yyyyMMdd"));
+                    logger.LogDebug($"{011111111111}:{length}");
+                }
+            });
+        }
+    }
+}

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.Traffic.Test/Program.cs

@@ -1,12 +1,43 @@
-using System;
+using JT808.DotNetty.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.DotNetty.Traffic.Test
 {
     class Program
     {
-        static void Main(string[] args)
+        async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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.AddSingleton<JT808DotNettyTrafficServiceTest>();
+                    services.AddJT808Configure()
+                                .AddJT808ClientKafka()
+                                .AddMsgConsumer(hostContext.Configuration)
+                                .AddInprocJT808Traffic();
+
+                    services.BuildServiceProvider().GetRequiredService<JT808DotNettyTrafficServiceTest>();
+                })
+                ;
+
+            await hostBuilder.RunConsoleAsync();
         }
     }
 }

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.Transmit.Test/JT808.DotNetty.Transmit.Test.csproj

@@ -5,4 +5,21 @@
     <TargetFramework>netcoreapp2.2</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Services\JT808.DotNetty.Transmit\JT808.DotNetty.Transmit.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.Transmit.Test/JT808DotNettyTransmitServiceTest.cs

@@ -0,0 +1,34 @@
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+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.DotNetty.Transmit.Test
+{
+    public class JT808DotNettyTransmitServiceTest
+    {
+        public readonly ILogger<JT808DotNettyTransmitServiceTest> logger;
+        public JT808DotNettyTransmitServiceTest(ILoggerFactory loggerFactory) {     
+            logger = loggerFactory.CreateLogger<JT808DotNettyTransmitServiceTest>();
+            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();
+                    }
+                }
+            });
+        }
+    }
+}

src/JT808.DotNetty.Services.Tests/JT808.DotNetty.Transmit.Test/Program.cs

@@ -1,12 +1,42 @@
-using System;
+using JT808.DotNetty.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.DotNetty.Transmit.Test
 {
     class Program
     {
-        static void Main(string[] args)
+        async static Task Main(string[] args)
         {
-            Console.WriteLine("Hello World!");
+            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.AddSingleton<JT808DotNettyTransmitServiceTest>();
+                    services.AddJT808Configure()
+                                .AddJT808ClientKafka()
+                                .AddMsgConsumer(hostContext.Configuration)
+                                .AddInprocJT808Transmit(hostContext.Configuration);
+                    services.BuildServiceProvider().GetRequiredService<JT808DotNettyTransmitServiceTest>();
+                })
+                ;
+
+            await hostBuilder.RunConsoleAsync();
         }
     }
 }

src/JT808.DotNetty.Services/JT808.DotNetty.MsgIdHandler/JT808.DotNetty.MsgIdHandler.csproj

@@ -16,6 +16,7 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\JT808.DotNetty.Abstractions\JT808.DotNetty.Abstractions.csproj" />
+    <ProjectReference Include="..\..\JT808.DotNetty.Kafka\JT808.DotNetty.Kafka.csproj" />
   </ItemGroup>
   <ItemGroup>
     <None Include="..\..\..\LICENSE">

src/JT808.DotNetty.Services/JT808.DotNetty.MsgIdHandler/IJT808DotNettyMsgIdHandlerExtensions.cs → src/JT808.DotNetty.Services/JT808.DotNetty.MsgIdHandler/JT808DotNettyMsgIdHandlerExtensions.cs

@@ -6,7 +6,7 @@ using System.Text;
 
 namespace JT808.DotNetty.MsgIdHandler
 {
-    public static class IJT808DotNettyMsgIdHandlerExtensions
+    public static class JT808DotNettyMsgIdHandlerExtensions
     {
         public static IJT808ClientBuilder AddJT808MsgIdHandler<TJT808DotNettyMsgIdHandler>(this IJT808ClientBuilder jT808ClientBuilder)
             where TJT808DotNettyMsgIdHandler: IJT808DotNettyMsgIdHandler

src/JT808.DotNetty.Services/JT808.DotNetty.ReplyMessage/JT808DotNettyReplyMessageExtensions.cs

@@ -28,7 +28,7 @@ namespace JT808.DotNetty.ReplyMessage
         public static IJT808ClientBuilder AddInprocJT808ReplyMessage<TReplyMessageService>(this IJT808ClientBuilder jT808ClientBuilder)
             where TReplyMessageService : JT808DotNettyReplyMessageService
         {
-            jT808ClientBuilder.JT808Builder.Services.AddSingleton<TReplyMessageService>();
+            jT808ClientBuilder.JT808Builder.Services.AddSingleton<JT808DotNettyReplyMessageService,TReplyMessageService>();
             jT808ClientBuilder.JT808Builder.Services.AddHostedService<JT808DotNettyReplyMessageHostedService>();
             return jT808ClientBuilder;
         }
@@ -41,7 +41,7 @@ namespace JT808.DotNetty.ReplyMessage
         public static IJT808ClientBuilder AddShareJT808ReplyMessage<TReplyMessageService>(this IJT808ClientBuilder jT808ClientBuilder)
             where TReplyMessageService : JT808DotNettyReplyMessageService
         {
-            jT808ClientBuilder.JT808Builder.Services.AddSingleton<TReplyMessageService>();
+            jT808ClientBuilder.JT808Builder.Services.AddSingleton<JT808DotNettyReplyMessageService, TReplyMessageService>();
             return jT808ClientBuilder;
         }
         /// <summary>

src/JT808.DotNetty.Services/JT808.DotNetty.SessionNotice/JT808DotNettySessionNoticeExtensions.cs

@@ -29,7 +29,7 @@ namespace JT808.DotNetty.SessionNotice
         public static IJT808ClientBuilder AddInprocJT808SessionNotice<TSessionNoticeService>(this IJT808ClientBuilder jT808ClientBuilder)
            where TSessionNoticeService : JT808DotNettySessionNoticeService
         {
-            jT808ClientBuilder.JT808Builder.Services.AddSingleton<TSessionNoticeService>();
+            jT808ClientBuilder.JT808Builder.Services.AddSingleton<JT808DotNettySessionNoticeService,TSessionNoticeService>();
             jT808ClientBuilder.JT808Builder.Services.AddHostedService<JT808DotNettySessionNoticeHostedService>();
             return jT808ClientBuilder;
         }
@@ -43,7 +43,7 @@ namespace JT808.DotNetty.SessionNotice
         public static IJT808ClientBuilder AddShareJT808SessionNotice<TSessionNoticeService>(this IJT808ClientBuilder jT808ClientBuilder)
           where TSessionNoticeService : JT808DotNettySessionNoticeService
         {
-            jT808ClientBuilder.JT808Builder.Services.AddSingleton<TSessionNoticeService>();
+            jT808ClientBuilder.JT808Builder.Services.AddSingleton<JT808DotNettySessionNoticeService, TSessionNoticeService>();
             return jT808ClientBuilder;
         }
 

src/JT808.DotNetty.Services/JT808.DotNetty.Traffic/JT808DotNettyTrafficServiceHostedService.cs

@@ -2,6 +2,7 @@
 using JT808.DotNetty.Abstractions;
 using Microsoft.Extensions.Hosting;
 using System.Threading;
+using JT808.Protocol.Extensions;
 
 namespace JT808.DotNetty.Traffic
 {
@@ -22,6 +23,7 @@ namespace JT808.DotNetty.Traffic
         {
             jT808MsgConsumer.Subscribe();
             jT808MsgConsumer.OnMessage((item)=> {
+                string str = item.Data.ToHexString();
                 jT808DotNettyTrafficService.Processor(item.TerminalNo, item.Data.Length);
             });
             return Task.CompletedTask;

src/JT808.DotNetty.Services/JT808.DotNetty.Transmit/JT808DotNettyTransmitHostedService.cs

@@ -1,22 +1,5 @@
-using DotNetty.Buffers;
-using DotNetty.Transport.Bootstrapping;
-using DotNetty.Transport.Channels;
-using DotNetty.Transport.Channels.Sockets;
-using System;
-using System.Collections.Generic;
-using System.Net;
-using System.Text;
-using System.Threading.Tasks;
-using DotNetty.Handlers.Logging;
-using Polly;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using JT808.DotNetty.Transmit.Configs;
-using System.Linq;
-using JT808.DotNetty.Transmit.Handlers;
+using System.Threading.Tasks;
 using JT808.DotNetty.Abstractions;
-using JT808.Protocol;
-using JT808.Protocol.Interfaces;
 using Microsoft.Extensions.Hosting;
 using System.Threading;
 
@@ -37,7 +20,7 @@ namespace JT808.DotNetty.Transmit
         public Task StartAsync(CancellationToken cancellationToken)
         {
             jT808MsgConsumer.Subscribe();
-            jT808MsgConsumer.OnMessage(jT808DotNettyTransmitService.SendAsync);
+            jT808MsgConsumer.OnMessage(jT808DotNettyTransmitService.Send);
             return Task.CompletedTask;
         }
 

src/JT808.DotNetty.Services/JT808.DotNetty.Transmit/JT808DotNettyTransmitService.cs

@@ -28,7 +28,7 @@ namespace JT808.DotNetty.Transmit
             this.optionsMonitor = optionsMonitor;
             InitialDispatcherClient();
         }
-        public void SendAsync((string TerminalNo, byte[] Data) parameter)
+        public void Send((string TerminalNo, byte[] Data) parameter)
         {
             if (optionsMonitor.CurrentValue.DataTransfer != null)
             {