8 changed files with 63 additions and 70 deletions
--- a/modules/JiShe.CollectBus.Kafka/AdminClient/AdminClientService.cs
+++ b/modules/JiShe.CollectBus.Kafka/AdminClient/AdminClientService.cs
@ -1,9 +1,7 @@
 using Confluent.Kafka;
 using Confluent.Kafka.Admin;
-using JiShe.CollectBus.Kafka.Internal;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
 using Volo.Abp.DependencyInjection;

 namespace JiShe.CollectBus.Kafka.AdminClient;
@ -11,17 +9,16 @@ namespace JiShe.CollectBus.Kafka.AdminClient;
 public class AdminClientService : IAdminClientService, IDisposable, ISingletonDependency
 {
    private readonly ILogger<AdminClientService> _logger;
-    private readonly KafkaOptionConfig _kafkaOptionConfig;
+
    /// <summary>
    /// Initializes a new instance of the <see cref="AdminClientService" /> class.
    /// </summary>
    /// <param name="configuration"></param>
    /// <param name="logger"></param>
-    public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig)
+    public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger)
    {
        _logger = logger;
-        _kafkaOptionConfig = kafkaOptionConfig.Value;
-        Instance = GetInstance();
+        Instance = GetInstance(configuration);
    }

    /// <summary>
@ -145,19 +142,22 @@ public class AdminClientService : IAdminClientService, IDisposable, ISingletonDe
    /// <summary>
    ///     Gets the instance.
    /// </summary>
+    /// <param name="configuration">The configuration.</param>
    /// <returns></returns>
-    public IAdminClient GetInstance()
+    public IAdminClient GetInstance(IConfiguration configuration)
    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(configuration["Kafka:EnableAuthorization"]);
+        var enableAuthorization = bool.Parse(configuration["Kafka:EnableAuthorization"]!);
        var adminClientConfig = new AdminClientConfig
        {
-            BootstrapServers = _kafkaOptionConfig.BootstrapServers
+            BootstrapServers = configuration["Kafka:BootstrapServers"]
        };
-        if (_kafkaOptionConfig.EnableAuthorization)
+        if (enableAuthorization)
        {
            adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
            adminClientConfig.SaslMechanism = SaslMechanism.Plain;
-            adminClientConfig.SaslUsername = _kafkaOptionConfig.SaslUserName;
-            adminClientConfig.SaslPassword = _kafkaOptionConfig.SaslUserName;
+            adminClientConfig.SaslUsername = configuration["Kafka:SaslUserName"];
+            adminClientConfig.SaslPassword = configuration["Kafka:SaslPassword"];
        }
        return new AdminClientBuilder(adminClientConfig).Build();
    }
--- a/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
+++ b/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
@ -34,8 +34,7 @@ namespace JiShe.CollectBus.Kafka.Internal
                ErrorCode.NotCoordinatorForGroup,
                ErrorCode.NetworkException,
                ErrorCode.GroupCoordinatorNotAvailable,
-                ErrorCode.InvalidGroupId,
-                ErrorCode.IllegalGeneration
+                ErrorCode.InvalidGroupId
            };
            return ex switch
            {
--- a/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs
+++ b/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs
@ -95,7 +95,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                DeviceId = $"{data.DeviceId}",
                DeviceType = $"{data.DeviceType.ToString()}.{IOTDBDataType.Data}",
                ProjectId = $"{data.ProjectId}",
-                Timestamps = data.TimeSpan!.Value.GetFormatTime(analysisBaseDto.DensityUnit, analysisBaseDto.TimeDensity).GetDateTimeOffset().ToUnixTimeNanoseconds(), 
+                Timestamps = data.TimeSpan!.Value.GetFormatTime(analysisBaseDto.DensityUnit, analysisBaseDto.TimeDensity).GetDateTimeOffset().ToUnixTimeMilliseconds(), 
                SingleMeasuring = new Tuple<string, T>(data.FiledName ?? string.Empty, data.DataValue ?? default)
            };
            _runtimeContext.UseTableSessionPool = true; // 使用表模型池
@ -198,7 +198,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                    DeviceId = $"{item.DeviceId}",
                    DeviceType = $"{item.DeviceType}.{IOTDBDataType.Data}",
                    ProjectId = $"{item.ProjectId}",
-                    Timestamps = item.TimeSpan!.Value.GetFormatTime(analysisBaseDto.DensityUnit, analysisBaseDto.TimeDensity).GetDateTimeOffset().ToUnixTimeNanoseconds(),  // TODO:这里暂时格式化15分钟数据，需要进行调整
+                    Timestamps = item.TimeSpan!.Value.GetFormatTime(analysisBaseDto.DensityUnit, analysisBaseDto.TimeDensity).GetDateTimeOffset().ToUnixTimeMilliseconds(),  // TODO:这里暂时格式化15分钟数据，需要进行调整
                    SingleMeasuring = new Tuple<string, T>(item.FiledName ?? string.Empty, item.DataValue ?? default)
                };
                _runtimeContext.UseTableSessionPool = true; // 使用表模型池
@ -275,7 +275,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
            if (!data.TimeSpan.HasValue)
                data.TimeSpan = analysisBaseDto.ReceivedTime;
            // 类型(心跳，登录，上电，掉电)
-            long timestamps = DateTime.Now.GetDateTimeOffset().ToUnixTimeNanoseconds();
+            long timestamps = DateTime.Now.GetDateTimeOffset().ToUnixTimeMilliseconds();
            var treeData = new TreeModelSingleMeasuringEntity<T>()
            {
                SystemName = _applicationOptions.SystemType,
@ -287,20 +287,6 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
            };
            _runtimeContext.UseTableSessionPool = false; // 使树模型池
            await _dbProvider.InsertAsync(treeData);
-            // 数据帧
-            var treeFrameData = new TreeModelSingleMeasuringEntity<string>()
-            {
-                SystemName = _applicationOptions.SystemType,
-                DeviceId = $"{data.DeviceId}",
-                DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
-                ProjectId = $"{data.ProjectId}",
-                Timestamps = timestamps,
-                SingleMeasuring = new Tuple<string, string>(ConcentratorStatusFieldConst.FrameData, analysisBaseDto.HexMessage ?? string.Empty)
-            };
-
-            _runtimeContext.UseTableSessionPool = false; // 使树模型池
-            await _dbProvider.InsertAsync(treeFrameData);
-
            // 时间
            var treeRecordingTimeData = new TreeModelSingleMeasuringEntity<long>()
            {
@ -309,7 +295,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
                ProjectId = $"{data.ProjectId}",
                Timestamps = timestamps,
-                SingleMeasuring = new Tuple<string, long>(ConcentratorStatusFieldConst.RecordingTime, (data.TimeSpan.HasValue ? data.TimeSpan.Value : DateTime.Now).GetDateTimeOffset().ToUnixTimeNanoseconds())
+                SingleMeasuring = new Tuple<string, long>(ConcentratorStatusFieldConst.RecordingTime, data.TimeSpan!.Value.GetDateTimeOffset().ToUnixTimeNanoseconds())
            };
            _runtimeContext.UseTableSessionPool = false; // 使树模型池
            await _dbProvider.InsertAsync(treeRecordingTimeData);
@ -325,7 +311,19 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
            };
            _runtimeContext.UseTableSessionPool = false; // 使树模型池
            await _dbProvider.InsertAsync(treeRemarkData);
-           
+            // 数据帧
+            var treeFrameData = new TreeModelSingleMeasuringEntity<string>()
+            {
+                SystemName = _applicationOptions.SystemType,
+                DeviceId = $"{data.DeviceId}",
+                DeviceType = $"{data.DeviceType}.{IOTDBDataType.Status}",
+                ProjectId = $"{data.ProjectId}",
+                Timestamps = timestamps,
+                SingleMeasuring = new Tuple<string, string>(ConcentratorStatusFieldConst.FrameData, analysisBaseDto.HexMessage ?? string.Empty )
+            };
+
+            _runtimeContext.UseTableSessionPool = false; // 使树模型池
+            await _dbProvider.InsertAsync(treeFrameData);
            return await Task.FromResult(true);
        }

--- a/protocols/JiShe.CollectBus.Protocol.T37612012/T37612012ProtocolPlugin.cs
+++ b/protocols/JiShe.CollectBus.Protocol.T37612012/T37612012ProtocolPlugin.cs
@ -170,7 +170,6 @@ namespace JiShe.CollectBus.Protocol.T37612012
            //    await _deviceRepository.UpdateAsync(entity);
            //}

-
            var messageReceivedLoginEvent = new MessageReceivedLogin
            {
                ClientId = code,
@ -181,8 +180,7 @@ namespace JiShe.CollectBus.Protocol.T37612012
                MessageId = Guid.NewGuid().ToString(),
                ReceivedTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")
            };
-            //await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent);
-
+            await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent);
            var reqParam = new ReqParameter2
            {
                AFN = AFN.确认或否认,
@ -213,7 +211,7 @@ namespace JiShe.CollectBus.Protocol.T37612012
            {
                await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
                _logger.LogInformation($"集中器地址{issuedEventMessage.ClientId} 登录回复下发内容:{Convert.ToHexString(bytes)}");
-               // await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, issuedEventMessage);
+                await _producerService.ProduceAsync(ProtocolConst.SubscriberLoginIssuedEventName, issuedEventMessage);
            }
        }

@ -268,7 +266,7 @@ namespace JiShe.CollectBus.Protocol.T37612012
                MessageId = Guid.NewGuid().ToString(),
                ReceivedTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")
            };
-           // await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent);
+            await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent);

            var reqParam = new ReqParameter2()
            {
@ -301,7 +299,7 @@ namespace JiShe.CollectBus.Protocol.T37612012
            {
                await _tcpService.SendAsync(issuedEventMessage.ClientId, bytes);
                _logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 心跳回复下发内容:{Convert.ToHexString(bytes)}");
-               // await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, issuedEventMessage);
+                await _producerService.ProduceAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, issuedEventMessage);
            }


--- a/services/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
+++ b/services/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
@ -38,7 +38,7 @@ namespace JiShe.CollectBus.Subscribers
        /// 电表自动阀控下行消息消费订阅
        /// </summary>
        /// <returns></returns>
-        Task<ISubscribeAck> AmmeterScheduledAutoValveControl(List<MeterReadingTelemetryPacketInfo> issuedEventMessage);
+        Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo issuedEventMessage);
        #endregion

        #region 水表消息采集
--- a/services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
+++ b/services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
@ -62,19 +62,21 @@ namespace JiShe.CollectBus.Subscribers
        [KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName, EnableBatch = true)]
        public async Task<ISubscribeAck> LoginIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
        {
-            bool isAck = true;
+            bool isAck = false;
            foreach (var issuedEventMessage in issuedEventMessages)
            {
-                //var loginEntity = await _messageReceivedLoginEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
-                //if (loginEntity == null)
-                //{
-                //    isAck=false;
-                //    break;
-                //}
+                var loginEntity = await _messageReceivedLoginEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
+                if (loginEntity == null)
+                {
+                    isAck=false;
+                    break;
+                }
                
-                //loginEntity.AckTime = Clock.Now;
-                //loginEntity.IsAck = true;
-                //await _messageReceivedLoginEventRepository.UpdateAsync(loginEntity);
+                loginEntity.AckTime = Clock.Now;
+                loginEntity.IsAck = true;
+                await _messageReceivedLoginEventRepository.UpdateAsync(loginEntity);
+               isAck = true;
+
            }
            // TODO:暂时ACK，等后续处理是否放到私信队列中
            return isAck? SubscribeAck.Success(): SubscribeAck.Fail();
@ -83,19 +85,19 @@ namespace JiShe.CollectBus.Subscribers
        [KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName, EnableBatch = true)]
        public async Task<ISubscribeAck> HeartbeatIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
        {
-            bool isAck = true;
-            //foreach (var issuedEventMessage in issuedEventMessages)
-            //{
-            //    var heartbeatEntity = await _messageReceivedHeartbeatEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
-            //    if (heartbeatEntity == null)
-            //    {
-            //        isAck = false;
-            //        break;
-            //    }
-            //    heartbeatEntity.AckTime = Clock.Now;
-            //    heartbeatEntity.IsAck = true;
-            //    await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
-            // }
+            bool isAck = false;
+            foreach (var issuedEventMessage in issuedEventMessages)
+            {
+                var heartbeatEntity = await _messageReceivedHeartbeatEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
+                if (heartbeatEntity == null)
+                {
+                    isAck = false;
+                    break;
+                }
+                heartbeatEntity.AckTime = Clock.Now;
+                heartbeatEntity.IsAck = true;
+                await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
+             }
            // TODO:暂时ACK，等后续处理是否放到私信队列中
            return isAck ? SubscribeAck.Success() : SubscribeAck.Fail();
        }
--- a/services/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
+++ b/services/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
@ -5,7 +5,6 @@ using JiShe.CollectBus.Kafka.Attributes;
 using JiShe.CollectBus.Kafka.Internal;
 using Microsoft.Extensions.Logging;
 using System;
-using System.Collections.Generic;
 using System.Threading.Tasks;
 using TouchSocket.Sockets;

@ -80,14 +79,12 @@ namespace JiShe.CollectBus.Subscribers
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
-        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName,EnableBatch =true,TaskCount=30,BatchSize =500)]
-        public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(List<MeterReadingTelemetryPacketInfo> receivedMessage)
+        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName)]
+        public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            //todo 如果是时段自动阀控，需要检查当前的时间，如果时间在自动阀控时间段内，则发送自动阀控报文，否则不发送，尤其是消息队列阻塞或者延时过长的时候。以免造成生产事故。
-            //_logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
-            _logger.LogWarning($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
-            return SubscribeAck.Success();
-            //return await SendMessagesAsync(receivedMessage);
+            _logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
+            return await SendMessagesAsync(receivedMessage);
        }

        /// <summary>
--- a/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml
+++ b/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml
@ -16,7 +16,6 @@
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet" />
    <title>后端服务</title>
-
 </head>
   
 <body>