Compare commits
No commits in common. "72d1b9f623ee392decfa8daafea9bd1f0367ee72" and "00372913500bf2c950fb5ce93dfabb7229fb12e9" have entirely different histories.
72d1b9f623
...
0037291350
@ -215,19 +215,11 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
|
|||||||
return aa == null;
|
return aa == null;
|
||||||
}
|
}
|
||||||
|
|
||||||
[KafkaSubscribe(["test-topic1"])]
|
[KafkaSubscribe(["test-topic"])]
|
||||||
|
|
||||||
public async Task<ISubscribeAck> KafkaSubscribeAsync() // TestSubscribe obj
|
public async Task<bool> KafkaSubscribeAsync(string obj)
|
||||||
{
|
{
|
||||||
var obj=string.Empty;
|
|
||||||
_logger.LogWarning($"收到订阅消息： {obj}");
|
_logger.LogWarning($"收到订阅消息： {obj}");
|
||||||
return SubscribeAck.Success();
|
return await Task.FromResult(true);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public class TestSubscribe
|
|
||||||
{
|
|
||||||
public string Topic { get; set; }
|
|
||||||
public int Val { get; set; }
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|||||||
@ -16,7 +16,6 @@
|
|||||||
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
|
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
|
||||||
<link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet"/>
|
<link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet"/>
|
||||||
<title>后端服务</title>
|
<title>后端服务</title>
|
||||||
|
|
||||||
</head>
|
</head>
|
||||||
|
|
||||||
<body>
|
<body>
|
||||||
|
|||||||
@ -1,21 +0,0 @@
|
|||||||
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; }
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,7 +1,4 @@
|
|||||||
using Confluent.Kafka;
|
using Confluent.Kafka;
|
||||||
using DeviceDetectorNET;
|
|
||||||
using JiShe.CollectBus.Common.Enums;
|
|
||||||
using JiShe.CollectBus.Common.Helpers;
|
|
||||||
using JiShe.CollectBus.Kafka.Attributes;
|
using JiShe.CollectBus.Kafka.Attributes;
|
||||||
using JiShe.CollectBus.Kafka.Consumer;
|
using JiShe.CollectBus.Kafka.Consumer;
|
||||||
using Microsoft.AspNetCore.Builder;
|
using Microsoft.AspNetCore.Builder;
|
||||||
@ -11,7 +8,6 @@ using Microsoft.Extensions.Logging;
|
|||||||
using Microsoft.Extensions.Primitives;
|
using Microsoft.Extensions.Primitives;
|
||||||
using Newtonsoft.Json;
|
using Newtonsoft.Json;
|
||||||
using System;
|
using System;
|
||||||
using System.Collections.Concurrent;
|
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
using System.Linq;
|
using System.Linq;
|
||||||
using System.Reflection;
|
using System.Reflection;
|
||||||
@ -100,7 +96,6 @@ namespace JiShe.CollectBus.Kafka
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// 处理消息
|
/// 处理消息
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@ -111,31 +106,25 @@ namespace JiShe.CollectBus.Kafka
|
|||||||
private static async Task<bool> ProcessMessageAsync(string message, MethodInfo method, object subscribe)
|
private static async Task<bool> ProcessMessageAsync(string message, MethodInfo method, object subscribe)
|
||||||
{
|
{
|
||||||
var parameters = method.GetParameters();
|
var parameters = method.GetParameters();
|
||||||
bool isGenericTask = method.ReturnType.IsGenericType
|
if (parameters.Length != 1)
|
||||||
&& method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
|
return true;
|
||||||
bool existParameters = parameters.Length > 0;
|
|
||||||
dynamic? messageObj= null;
|
|
||||||
if (existParameters)
|
|
||||||
{
|
|
||||||
var paramType = parameters[0].ParameterType;
|
var paramType = parameters[0].ParameterType;
|
||||||
messageObj = paramType == typeof(string) ? message : message.Deserialize(paramType);
|
var messageObj = paramType == typeof(string)? message: JsonConvert.DeserializeObject(message, paramType);
|
||||||
}
|
|
||||||
if (isGenericTask)
|
if (method.ReturnType == typeof(Task))
|
||||||
{
|
{
|
||||||
object? result = await (Task<ISubscribeAck>)method.Invoke(subscribe, existParameters? new[] { messageObj }:null)!;
|
object? result = await (Task<bool>)method.Invoke(subscribe, new[] { messageObj })!;
|
||||||
if (result is ISubscribeAck ackResult)
|
if (result is bool success)
|
||||||
{
|
return success;
|
||||||
return ackResult.Ack;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
object? result = method.Invoke(subscribe, existParameters ? new[] { messageObj } : null);
|
object? result = method.Invoke(subscribe, new[] { messageObj });
|
||||||
if (result is ISubscribeAck ackResult)
|
if (result is bool success)
|
||||||
{
|
return success;
|
||||||
return ackResult.Ack;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -1,75 +0,0 @@
|
|||||||
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);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
Loading…
x
Reference in New Issue
Block a user