60 lines
2.2 KiB
C#
60 lines
2.2 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Threading.Tasks;
|
|
using JiShe.CollectBus.IotSystems.MessageReceiveds;
|
|
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
|
|
using JiShe.CollectBus.Protocol.Contracts.Models;
|
|
using MassTransit;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using Volo.Abp.Domain.Repositories;
|
|
|
|
namespace JiShe.CollectBus.Consumers
|
|
{
|
|
/// <summary>
|
|
/// Batch 一次最多 100 个，最多 10 个并发批次
|
|
/// </summary>
|
|
public class ReceivedConsumer: IConsumer<Batch<MessageReceived>>
|
|
{
|
|
private readonly ILogger<ReceivedConsumer> _logger;
|
|
private readonly IServiceProvider _serviceProvider;
|
|
private readonly IRepository<MessageReceived, Guid> _messageReceivedEventRepository;
|
|
|
|
/// <summary>
|
|
/// MessageReceivedConsumer
|
|
/// </summary>
|
|
/// <param name="logger"></param>
|
|
/// <param name="serviceProvider"></param>
|
|
/// <param name="messageReceivedEventRepository"></param>
|
|
public ReceivedConsumer(ILogger<ReceivedConsumer> logger,
|
|
IServiceProvider serviceProvider,
|
|
IRepository<MessageReceived, Guid> messageReceivedEventRepository)
|
|
{
|
|
_logger = logger;
|
|
_serviceProvider = serviceProvider;
|
|
_messageReceivedEventRepository = messageReceivedEventRepository;
|
|
}
|
|
|
|
|
|
public async Task Consume(ConsumeContext<Batch<MessageReceived>> context)
|
|
{
|
|
const string protocolType = "Standard";
|
|
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>(protocolType);
|
|
if (protocolPlugin == null)
|
|
{
|
|
_logger.LogError("协议不存在！");
|
|
}
|
|
else
|
|
{
|
|
var list = new List<MessageReceived>();
|
|
foreach (var contextItem in context.Message)
|
|
{
|
|
await protocolPlugin.AnalyzeAsync<TB3761>(contextItem.Message);
|
|
list.Add(contextItem.Message);
|
|
}
|
|
await _messageReceivedEventRepository.InsertManyAsync(list);
|
|
}
|
|
}
|
|
}
|
|
}
|