JiShe.CollectBus/JiShe.CollectBus.RabbitMQ/Consumers/MessageIssuedConsumer.cs

using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.MongoDB;
using JiShe.CollectBus.Protocol.Contracts.Models;
using MassTransit;
using Microsoft.Extensions.Logging;
using TouchSocket.Sockets;

namespace JiShe.CollectBus.RabbitMQ.Consumers
{
    public class MessageIssuedConsumer(
        ILogger<MessageIssuedEvent> logger,
        ITcpService tcpService,
        IMongoRepository<MessageReceivedHeartbeatEvent> mongoHeartbeatRepository,
        IMongoRepository<MessageReceivedLoginEvent> mongoLoginRepository)
        : IConsumer<MessageIssuedEvent>
    {
        public async Task Consume(ConsumeContext<MessageIssuedEvent> context)
        {
            switch (context.Message.Type)
            {
                case IssuedEventType.Heartbeat:
                    await mongoHeartbeatRepository.UpdateAsync(a => a.MessageId == context.Message.MessageId, b => new MessageReceivedHeartbeatEvent { IsAck = true, AckTime = DateTime.Now });
                    break;
                case IssuedEventType.Login:
                    await mongoLoginRepository.UpdateAsync(a => a.MessageId == context.Message.MessageId, b => new MessageReceivedLoginEvent { IsAck = true, AckTime = DateTime.Now });
                    break;
                case IssuedEventType.Data:
                    break;
                default:
                    throw new ArgumentOutOfRangeException();
            }
            await tcpService.SendAsync(context.Message.ClientId, context.Message.Message);
        }
    }
}