优化订阅回调

暂存
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)
+            bool isGenericTask = method.ReturnType.IsGenericType
-                return true;
+                  && method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
-
+            bool existParameters = parameters.Length > 0;
-            var paramType = parameters[0].ParameterType;
+            dynamic? messageObj= null;
-            var messageObj = paramType == typeof(string)? message: JsonConvert.DeserializeObject(message, paramType);
+            if (existParameters)
            if (method.ReturnType == typeof(Task))
            {
-                object? result = await (Task<bool>)method.Invoke(subscribe, new[] { messageObj })!;
+                 var paramType = parameters[0].ParameterType;
-                if (result is bool success)
+                 messageObj = paramType == typeof(string) ? message : message.Deserialize(paramType);
-                    return success;
+            }
            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 });
+                object? result = method.Invoke(subscribe, existParameters ? new[] { messageObj } : null);
-                if (result is bool success)
+                if (result is ISubscribeAck ackResult)
-                   return success;
+                {
                    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