40 lines
1.4 KiB
C#
40 lines
1.4 KiB
C#
using JiShe.CollectBus.Common.Models;
|
|
using JiShe.CollectBus.MongoDB;
|
|
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
|
|
using MassTransit;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace JiShe.CollectBus.MQ.Consumer
|
|
{
|
|
/// <summary>
|
|
/// Batch 一次最多 100 个，最多 10 个并发批次
|
|
/// </summary>
|
|
public class MessageReceivedConsumer(
|
|
ILogger<MessageReceivedConsumer> logger,
|
|
IServiceProvider serviceProvider,
|
|
IMongoRepository<MessageReceivedEvent> mongoReceivedRepository)
|
|
: IConsumer<Batch<MessageReceivedEvent>>
|
|
{
|
|
public async Task Consume(ConsumeContext<Batch<MessageReceivedEvent>> context)
|
|
{
|
|
const string protocolType = "Standard";
|
|
var protocolPlugin = serviceProvider.GetKeyedService<IProtocolPlugin>(protocolType);
|
|
if (protocolPlugin == null)
|
|
{
|
|
logger.LogError("协议不存在！");
|
|
}
|
|
else
|
|
{
|
|
var list = new List<MessageReceivedEvent>();
|
|
foreach (var contextItem in context.Message)
|
|
{
|
|
await protocolPlugin.AnalyzeAsync(contextItem.Message);
|
|
list.Add(contextItem.Message);
|
|
}
|
|
await mongoReceivedRepository.InsertManyAsync(list);
|
|
}
|
|
}
|
|
}
|
|
}
|