优化订阅回调

暂存
2025-04-15 18:03:51 +08:00 · 2025-04-15 16:45:10 +08:00
5 changed files with 133 additions and 17 deletions
--- a/src/JiShe.CollectBus.Application/Samples/SampleAppService.cs
+++ b/src/JiShe.CollectBus.Application/Samples/SampleAppService.cs
@ -215,11 +215,19 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
        return aa == null;
    }

-    [KafkaSubscribe(["test-topic"])]
+    [KafkaSubscribe(["test-topic1"])]

-    public async Task<bool> KafkaSubscribeAsync(string obj)
+    public async Task<ISubscribeAck> KafkaSubscribeAsync() // TestSubscribe obj
    {
+        var obj=string.Empty;
        _logger.LogWarning($"收到订阅消息： {obj}");
-       return await Task.FromResult(true);
+       return SubscribeAck.Success();
    }
 }
+
+public class TestSubscribe 
+{
+    public string Topic { get; set; }
+    public int Val { get; set; }
+}
+
--- a/src/JiShe.CollectBus.Host/Pages/Monitor.cshtml
+++ b/src/JiShe.CollectBus.Host/Pages/Monitor.cshtml
@ -16,6 +16,7 @@
    <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>
--- a/src/JiShe.CollectBus.KafkaProducer/ISubscribeAck.cs
+++ b/src/JiShe.CollectBus.KafkaProducer/ISubscribeAck.cs
@ -0,0 +1,21 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Kafka
+{
+    public interface ISubscribeAck
+    {
+        /// <summary>
+        /// 是否成功标记
+        /// </summary>
+        bool Ack { get; set; }
+
+        /// <summary>
+        /// 消息
+        /// </summary>
+        string? Msg { get; set; }
+    }
+}
--- a/src/JiShe.CollectBus.KafkaProducer/KafkaSubcribesExtensions.cs
+++ b/src/JiShe.CollectBus.KafkaProducer/KafkaSubcribesExtensions.cs
@ -1,4 +1,7 @@
 using Confluent.Kafka;
+using DeviceDetectorNET;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.Common.Helpers;
 using JiShe.CollectBus.Kafka.Attributes;
 using JiShe.CollectBus.Kafka.Consumer;
 using Microsoft.AspNetCore.Builder;
@ -8,6 +11,7 @@ using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Primitives;
 using Newtonsoft.Json;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
@ -96,6 +100,7 @@ namespace JiShe.CollectBus.Kafka
            });
        }

+
        /// <summary>
        /// 处理消息
        /// </summary>
@ -106,25 +111,31 @@ namespace JiShe.CollectBus.Kafka
        private static async Task<bool> ProcessMessageAsync(string message, MethodInfo method, object subscribe)
        {
            var parameters = method.GetParameters();
-            if (parameters.Length != 1)
-                return true;
-
-            var paramType = parameters[0].ParameterType;
-            var messageObj = paramType == typeof(string)? message: JsonConvert.DeserializeObject(message, paramType);
-
-            if (method.ReturnType == typeof(Task))
+            bool isGenericTask = method.ReturnType.IsGenericType
+                  && method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
+            bool existParameters = parameters.Length > 0;
+            dynamic? messageObj= null;
+            if (existParameters)
            {
-                object? result = await (Task<bool>)method.Invoke(subscribe, new[] { messageObj })!;
-                if (result is bool success)
-                    return success;
+                 var paramType = parameters[0].ParameterType;
+                 messageObj = paramType == typeof(string) ? message : message.Deserialize(paramType);
+            }
+            if (isGenericTask)
+            {
+                object? result = await (Task<ISubscribeAck>)method.Invoke(subscribe, existParameters? new[] { messageObj }:null)!;
+                if (result is ISubscribeAck ackResult)
+                {
+                    return ackResult.Ack;
+                }
            }
            else
            {
-                object? result = method.Invoke(subscribe, new[] { messageObj });
-                if (result is bool success)
-                   return success;
+                object? result = method.Invoke(subscribe, existParameters ? new[] { messageObj } : null);
+                if (result is ISubscribeAck ackResult)
+                {
+                    return ackResult.Ack;
+                }
            }
-
            return false;
        }

--- a/src/JiShe.CollectBus.KafkaProducer/SubscribeResult.cs
+++ b/src/JiShe.CollectBus.KafkaProducer/SubscribeResult.cs
@ -0,0 +1,75 @@
+using Confluent.Kafka;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using static System.Runtime.InteropServices.JavaScript.JSType;
+
+namespace JiShe.CollectBus.Kafka
+{
+    public class SubscribeResult: ISubscribeAck
+    {
+        /// <summary>
+        /// 是否成功
+        /// </summary>
+        public bool Ack { get; set; }
+
+        /// <summary>
+        /// 消息
+        /// </summary>
+        public string? Msg { get; set; }
+
+
+        /// <summary>
+        /// 成功
+        /// </summary>
+        /// <param name="msg">消息</param>
+        public SubscribeResult Success(string? msg = null)
+        {
+            Ack = true;
+            Msg = msg;
+            return this;
+        }
+
+        /// <summary>
+        /// 失败
+        /// </summary>
+        /// <param name="code"></param>
+        /// <param name="msg"></param>
+        /// <param name="data"></param>
+        /// <returns></returns>
+        public SubscribeResult Fail(string? msg = null)
+        {
+            Msg = msg;
+            Ack = false;
+            return this;
+        }
+    }
+
+    public static partial class SubscribeAck
+    {
+
+        /// <summary>
+        /// 成功
+        /// </summary>
+        /// <param name="msg">消息</param>
+        /// <returns></returns>
+        public static ISubscribeAck Success(string? msg = null)
+        {
+            return new SubscribeResult().Success(msg);
+        }
+
+
+        /// <summary>
+        /// 失败
+        /// </summary>
+        /// <param name="msg">消息</param>
+        /// <returns></returns>
+        public static ISubscribeAck Fail(string? msg = null)
+        {
+            return new SubscribeResult().Fail(msg);
+        }
+    }
+
+}
Author	SHA1	Message	Date
zenghongyao	72d1b9f623	优化订阅回调	2025-04-15 18:03:51 +08:00
zenghongyao	7b1f04a452	暂存	2025-04-15 16:45:10 +08:00