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;
|
||||
}
|
||||
|
||||
[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}");
|
||||
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"/>
|
||||
<link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet"/>
|
||||
<title>后端服务</title>
|
||||
|
||||
</head>
|
||||
|
||||
<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 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;
|
||||
@ -11,7 +8,6 @@ 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;
|
||||
@ -100,7 +96,6 @@ namespace JiShe.CollectBus.Kafka
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 处理消息
|
||||
/// </summary>
|
||||
@ -111,31 +106,25 @@ namespace JiShe.CollectBus.Kafka
|
||||
private static async Task<bool> ProcessMessageAsync(string message, MethodInfo method, object subscribe)
|
||||
{
|
||||
var parameters = method.GetParameters();
|
||||
bool isGenericTask = method.ReturnType.IsGenericType
|
||||
&& method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
|
||||
bool existParameters = parameters.Length > 0;
|
||||
dynamic? messageObj= null;
|
||||
if (existParameters)
|
||||
{
|
||||
if (parameters.Length != 1)
|
||||
return true;
|
||||
|
||||
var paramType = parameters[0].ParameterType;
|
||||
messageObj = paramType == typeof(string) ? message : message.Deserialize(paramType);
|
||||
}
|
||||
if (isGenericTask)
|
||||
var messageObj = paramType == typeof(string)? message: JsonConvert.DeserializeObject(message, paramType);
|
||||
|
||||
if (method.ReturnType == typeof(Task))
|
||||
{
|
||||
object? result = await (Task<ISubscribeAck>)method.Invoke(subscribe, existParameters? new[] { messageObj }:null)!;
|
||||
if (result is ISubscribeAck ackResult)
|
||||
{
|
||||
return ackResult.Ack;
|
||||
}
|
||||
object? result = await (Task<bool>)method.Invoke(subscribe, new[] { messageObj })!;
|
||||
if (result is bool success)
|
||||
return success;
|
||||
}
|
||||
else
|
||||
{
|
||||
object? result = method.Invoke(subscribe, existParameters ? new[] { messageObj } : null);
|
||||
if (result is ISubscribeAck ackResult)
|
||||
{
|
||||
return ackResult.Ack;
|
||||
}
|
||||
object? result = method.Invoke(subscribe, new[] { messageObj });
|
||||
if (result is bool success)
|
||||
return success;
|
||||
}
|
||||
|
||||
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