添加生产消费配置

src/JT809.DotNetty.Core/Internal/JT809SubordinateLinkNotifyImplService.cs

@@ -44,18 +44,17 @@ namespace JT809.DotNetty.Core.Internal
         {
             if (configuration.SubordinateClientEnable)
             {
-#warning JT809GlobalConfig 
+#warning jT809SuperiorMainSessionManager
                 //var session = jT809SuperiorMainSessionManager.GetSession(JT809GlobalConfig.Instance.HeaderOptions.MsgGNSSCENTERID);
                 //if (session != null)
                 //{
                 //    //发送从链路注销请求
                 //    var package = JT809BusinessType.从链路断开通知消息.Create(new JT809_0x9007()
                 //    {
-#warning JT809_0x9007_ReasonCode???
-                //        ErrorCode =  JT809_0x1007_ErrorCode.主链路断开
+                //        ReasonCode   = reasonCode
                 //    });
                 //    JT809Response jT809Response = new JT809Response(package, 100);
-                //    if(logger.IsEnabled(LogLevel.Information))
+                //    if (logger.IsEnabled(LogLevel.Information))
                 //        logger.LogInformation($"从链路断开通知消息>>>{JT809Serializer.Serialize(package, 100).ToHexString()}");
                 //    session.Channel.WriteAndFlushAsync(jT809Response);
                 //}

src/JT809.DotNetty.Core/JT809.DotNetty.Core.csproj

@@ -26,7 +26,7 @@
     <PackageReference Include="DotNetty.Transport.Libuv" Version="0.6.0" />
     <PackageReference Include="DotNetty.Codecs" Version="0.6.0" />
     <PackageReference Include="DotNetty.Codecs.Http" Version="0.6.0" />
-    <PackageReference Include="JT809" Version="2.1.4-preview3" />
+    <PackageReference Include="JT809" Version="2.1.4-preview4" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.4" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.1.4" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />

src/JT809.DotNetty.Simples/Inferior/JT809.Inferior.Client/JT809.Inferior.Client.csproj

@@ -6,7 +6,7 @@
     <LangVersion>7.3</LangVersion>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="JT809" Version="2.1.4-preview3" />
+    <PackageReference Include="JT809" Version="2.1.4-preview4" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.1.4" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.4" />
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.1.4" />

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/Configs/JT809ConsumerConfig.cs

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

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/Configs/JT809ProducerConfig.cs

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

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809Consumer.cs

@@ -1,5 +1,5 @@
 using Confluent.Kafka;
-using JT809.PubSub.Abstractions;
+using JT809.KafkaService.Configs;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using System;
@@ -16,21 +16,20 @@ namespace JT809.KafkaService
 
         protected ILogger logger { get; }
 
-        protected override IList<IConsumer<string, T>> Consumers { get; }
+        protected override IConsumer<string, T> Consumer { get; }
 
         protected JT809Consumer(
-            IOptions<ConsumerConfig> consumerConfigAccessor,
+            IOptions<JT809ConsumerConfig> consumerConfigAccessor,
             ILoggerFactory loggerFactory) 
             : base(consumerConfigAccessor.Value)
         {
             logger = loggerFactory.CreateLogger("JT809Consumer");
-            Consumers = new List<IConsumer<string, T>>();
             ConsumerBuilder<string, T> consumerBuilder = new ConsumerBuilder<string, T>(ConsumerConfig);
             consumerBuilder.SetErrorHandler((consumer, error) =>
             {
                 logger.LogError(error.Reason);
             });
-            Consumers.Add(consumerBuilder.Build());
+            Consumer = consumerBuilder.Build();
         }
 
         public override void OnMessage(Action<(string MsgId, T Data)> callback)
@@ -44,7 +43,7 @@ namespace JT809.KafkaService
                         //如果不指定分区，根据kafka的机制会从多个分区中拉取数据
                         //如果指定分区，根据kafka的机制会从相应的分区中拉取数据
                         //consumers[n].Assign(topicPartitionList[n]);
-                        var data = Consumers[0].Consume(Cts.Token);
+                        var data = Consumer.Consume(Cts.Token);
                         if (logger.IsEnabled(LogLevel.Debug))
                         {
                             logger.LogDebug($"Topic: {data.Topic} Key: {data.Message.Key} Partition: {data.Partition} Offset: {data.Offset} Data:{string.Join("", data.Message.Value)} TopicPartitionOffset:{data.TopicPartitionOffset}");
@@ -53,13 +52,11 @@ namespace JT809.KafkaService
                     }
                     catch (ConsumeException ex)
                     {
-#warning topicname
-                        logger.LogError(ex, "");
+                        logger.LogError(ex, ConsumerConfig.TopicName);
                     }
                     catch (Exception ex)
                     {
-#warning topicname
-                        logger.LogError(ex, "");
+                        logger.LogError(ex, ConsumerConfig.TopicName);
                     }
                 }
             }, Cts.Token);            
@@ -69,14 +66,13 @@ namespace JT809.KafkaService
         {
             if (_disposed) return;
             //仅有一个分区才需要订阅
-#warning topicname
-            Consumers[0].Subscribe("");
+            Consumer.Subscribe(ConsumerConfig.TopicName);
         }
 
         public override void Unsubscribe()
         {
             if (_disposed) return;
-            Consumers[0].Unsubscribe(); 
+            Consumer.Unsubscribe(); 
         }
 
         public override void Dispose()
@@ -97,8 +93,8 @@ namespace JT809.KafkaService
             if (disposing)
             {
                 Cts.Cancel();
-                Consumers[0].Close();
-                Consumers[0].Dispose();
+                Consumer.Close();
+                Consumer.Dispose();
                 Cts.Dispose();
             }
             _disposed = true;

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809ConsumerBase.cs

@@ -1,4 +1,5 @@
 using Confluent.Kafka;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -11,15 +12,15 @@ namespace JT809.KafkaService
 {
     public abstract class JT809ConsumerBase<T> : IJT808ConsumerOfT<T>
     {
-        public ConsumerConfig ConsumerConfig { get; }
+        public JT809ConsumerConfig ConsumerConfig { get; }
 
-        protected JT809ConsumerBase( ConsumerConfig config)
+        protected JT809ConsumerBase(IOptions<JT809ConsumerConfig> config)
         {
-            ConsumerConfig = config;
+            ConsumerConfig = config.Value;
         }
 
         public abstract CancellationTokenSource Cts { get; }
-        protected abstract IList<IConsumer<string, T>> Consumers { get; }
+        protected abstract IConsumer<string, T> Consumer { get; }
 
         public abstract void Dispose();
         public abstract void OnMessage(Action<(string MsgId, T Data)> callback);

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809Producer.cs

@@ -1,5 +1,6 @@
 using Confluent.Kafka;
 using Confluent.Kafka.Admin;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Options;
 using System;
@@ -25,7 +26,7 @@ namespace JT809.KafkaService
         protected override IProducer<string, T> Producer { get; }
 
         protected JT809Producer(
-            IOptions<ProducerConfig> producerConfigAccessor)
+            IOptions<JT809ProducerConfig> producerConfigAccessor)
             : base(producerConfigAccessor.Value)
         {
             Producer = CreateProducer();
@@ -51,8 +52,7 @@ namespace JT809.KafkaService
         public override async void ProduceAsync(string msgId, string vno_color, T data)
         {
             if (_disposed) return;
-#warning topicname
-            await Producer.ProduceAsync("", new Message<string, T>
+            await Producer.ProduceAsync(ProducerConfig.TopicName, new Message<string, T>
             {
                 Key = msgId,
                 Value = data

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809ProducerBase.cs

@@ -1,5 +1,7 @@
 using Confluent.Kafka;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
+using Microsoft.Extensions.Options;
 using System;
 using System.Collections.Generic;
 using System.Text;
@@ -8,12 +10,12 @@ namespace JT809.KafkaService
 {
     public abstract class JT809ProducerBase<T> : IJT809ProducerOfT<T>
     {
-        protected JT809ProducerBase(ProducerConfig config)
+        protected JT809ProducerBase(IOptions<JT809ProducerConfig> config)
         {
-            ProducerConfig = config;
+            ProducerConfig = config.Value;
         }
 
-        public ProducerConfig ProducerConfig { get;}
+        public JT809ProducerConfig ProducerConfig { get;}
         protected abstract IProducer<string, T> Producer { get;}
         public abstract void Dispose();
         public abstract void ProduceAsync(string msgId, string vno_color, T data);

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809_GpsPositio_Producer.cs

@@ -1,6 +1,7 @@
 using Confluent.Kafka;
 using Google.Protobuf;
 using JT809.GrpcProtos;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Options;
 
@@ -8,7 +9,7 @@ namespace JT809.KafkaService
 {
     public sealed class JT809_GpsPositio_Producer : JT809Producer<JT809GpsPosition>
     {
-        public JT809_GpsPositio_Producer(IOptions<ProducerConfig> producerConfigAccessor) : base( producerConfigAccessor)
+        public JT809_GpsPositio_Producer(IOptions<JT809ProducerConfig> producerConfigAccessor) : base( producerConfigAccessor)
         {
         }
     }

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809_GpsPosition_Consumer.cs

@@ -1,6 +1,7 @@
 using Confluent.Kafka;
 using Google.Protobuf;
 using JT809.GrpcProtos;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -13,7 +14,7 @@ namespace JT809.KafkaService
 {
     public sealed class JT809_GpsPosition_Consumer : JT809Consumer<JT809GpsPosition>
     {
-        public JT809_GpsPosition_Consumer(IOptions<ConsumerConfig> consumerConfigAccessor, ILoggerFactory loggerFactory) : base( consumerConfigAccessor, loggerFactory)
+        public JT809_GpsPosition_Consumer(IOptions<JT809ConsumerConfig> consumerConfigAccessor, ILoggerFactory loggerFactory) : base( consumerConfigAccessor, loggerFactory)
         {
         }
     }

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809_Same_Consumer.cs

@@ -1,6 +1,7 @@
 using Confluent.Kafka;
 using Google.Protobuf;
 using JT809.GrpcProtos;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -13,7 +14,7 @@ namespace JT809.KafkaService
 {
     public sealed class JT809_Same_Consumer : JT809Consumer<byte[]>
     {
-        public JT809_Same_Consumer(IOptions<ConsumerConfig> consumerConfigAccessor, ILoggerFactory loggerFactory) 
+        public JT809_Same_Consumer(IOptions<JT809ConsumerConfig> consumerConfigAccessor, ILoggerFactory loggerFactory) 
             : base(consumerConfigAccessor, loggerFactory)
         {
         }

src/JT809.DotNetty.Simples/Superior/JT809.KafkaService/JT809_Same_Producer.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Text;
 using Confluent.Kafka;
+using JT809.KafkaService.Configs;
 using JT809.PubSub.Abstractions;
 using Microsoft.Extensions.Options;
 
@@ -9,7 +10,7 @@ namespace JT809.KafkaService
 {
     public sealed class JT809_Same_Producer : JT809Producer<byte[]>
     {
-        public JT809_Same_Producer(IOptions<ProducerConfig> producerConfigAccessor)
+        public JT809_Same_Producer(IOptions<JT809ProducerConfig> producerConfigAccessor)
             : base(producerConfigAccessor)
         {
         }

src/JT809.DotNetty.Tests/JT809.DotNetty.Host.Test/JT809.DotNetty.Host.Test.csproj

@@ -6,7 +6,7 @@
     <LangVersion>7.3</LangVersion>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="JT809" Version="2.1.4-preview3" />
+    <PackageReference Include="JT809" Version="2.1.4-preview4" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="3.1.4" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.4" />
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.1.4" />