调整解析器

HealthMonitor.Service/Sub/MsgQueueManager.cs

@@ -1,5 +1,6 @@
 using HealthMonitor.Service.Resolver;
 using HealthMonitor.Service.Resolver.Interface;
+using HealthMonitor.Service.Sub.Topic.Model;
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Concurrent;
@@ -13,11 +14,16 @@ namespace HealthMonitor.Service.Sub
     public class MsgQueueManager: ConcurrentQueue<PackageMsgModel>
     {
         private readonly ILogger<MsgQueueManager> _logger;
-        private readonly BloodpressResolver _resolver;
-        public MsgQueueManager(ILogger<MsgQueueManager> logger, BloodpressResolver resolver) 
+        private readonly BloodpressResolver _resolverBloodpress;
+
+        //private const string BP = nameof(TopicHmBloodPress).ToLower();
+        
+
+
+        public MsgQueueManager(ILogger<MsgQueueManager> logger, BloodpressResolver resolverBloodpress) 
         { 
             _logger = logger;
-            _resolver = resolver;
+            _resolverBloodpress = resolverBloodpress;
         }
 
         //public IResolver? GetMsgResolver()
@@ -34,14 +40,24 @@ namespace HealthMonitor.Service.Sub
 
         public IResolver? GetMsgResolver(PackageMsgModel msg)
         {
-            //TryDequeue(out var msg);
-            if (msg == null)
+            if (msg == null) return null;
+            //switch (msg.Topic)
+            //{
+            //    case BP:
+            //        _resolverBloodpress.SetResolveInfo(msg);
+            //        return _resolverBloodpress;
+            //    default:
+            //        return null;
+            //}
+
+            if (msg.Topic.Equals(nameof(TopicHmBloodPress).ToLower()))
             {
-                return null;
+                _resolverBloodpress.SetResolveInfo(msg);
+                return _resolverBloodpress;
             }
-            _resolver.SetResolveInfo(msg);
+            return null;
 
-            return _resolver;
+            
         }
     }
 }

HealthMonitor.Service/Sub/TDengineDataSubcribe.cs

@@ -37,7 +37,6 @@ namespace HealthMonitor.Service.Sub
         //private int cnt = 0;
 
         public TDengineDataSubcribe(
-
             TDengineService serviceTDengine,
             PersonCacheManager personCacheMgr,
             BloodPressReferenceValueCacheManager bpRefValCacheManager,
@@ -109,14 +108,7 @@ namespace HealthMonitor.Service.Sub
 
         public void DoReceive(IntPtr Connection)
         {
-            var cfg = new ConsumerConfig
-            {
-                GourpId = "group_1",
-                TDConnectUser = _configTDengineService.UserName,
-                TDConnectPasswd = _configTDengineService.Password,
-                MsgWithTableName = "true",
-                TDConnectIp = _configTDengineService.Host,
-            };
+            
            // string topic = "topic_hm_bp_stats";
             string topic = nameof(TopicHmBloodPress).ToLower();
             TopicHmBloodPress fields = new();
@@ -142,7 +134,14 @@ namespace HealthMonitor.Service.Sub
                 _logger.LogError($"create topic failed, reason:{TDengine.Error(res)}");
                 throw new Exception($"create topic failed, reason:{TDengine.Error(res)}");
             }
-
+            var cfg = new ConsumerConfig
+            {
+                GourpId = "group_1",
+                TDConnectUser = _configTDengineService.UserName,
+                TDConnectPasswd = _configTDengineService.Password,
+                MsgWithTableName = "true",
+                TDConnectIp = _configTDengineService.Host,
+            };
             // create consumer 
             var consumer = new ConsumerBuilder(cfg)
                 .Build();
@@ -155,7 +154,6 @@ namespace HealthMonitor.Service.Sub
                 var consumeRes = consumer.Consume(300);
                 foreach (KeyValuePair<TopicPartition, TaosResult> kv in consumeRes.Message)
                 {
-
                     for (int i = 0; i < kv.Value.Datas.Count; i++)
                     {
                         if (((i + 1) % kv.Value.Metas.Count == 0))
@@ -186,8 +184,7 @@ namespace HealthMonitor.Service.Sub
                                
                             }
                         }
-                    }
-                      
+                    }  
                 }
                 consumer.Commit(consumeRes);
                 //_logger.LogInformation("监听中....");