50 lines
2.0 KiB
C#
50 lines
2.0 KiB
C#
using IM.Commons.IntegrationEvents;
|
|
using MassTransit;
|
|
using MediatR;
|
|
using MessageService.Domain.Events;
|
|
using MessageService.Domain.IReposities;
|
|
using MessageService.Infrastructure;
|
|
|
|
namespace MessageService.WebApi.Application.EventHandlers
|
|
{
|
|
public class MessageHandler : INotificationHandler<MessageCreatedDomainEvent>, INotificationHandler<MessageWithdrawDomainEvent>
|
|
{
|
|
private readonly IPublishEndpoint endpoint;
|
|
private readonly IConversationReposity reposity;
|
|
private readonly MessageDbContext messageDb;
|
|
|
|
public MessageHandler(IPublishEndpoint endpoint, IConversationReposity reposity, MessageDbContext messageDb)
|
|
{
|
|
this.endpoint = endpoint;
|
|
this.reposity = reposity;
|
|
this.messageDb = messageDb;
|
|
}
|
|
|
|
public async Task Handle(MessageCreatedDomainEvent notification, CancellationToken cancellationToken)
|
|
{
|
|
var message = notification.Message;
|
|
|
|
if(message.ChatType == Domain.Enums.ChatType.PRIVATE)
|
|
{
|
|
var list = await reposity.FindByStreamKeyAsync(message.StreamKey);
|
|
var owner = list.First(x => x.UserId == message.SenderId);
|
|
var target = list.First(x => x.UserId == message.TargetId);
|
|
|
|
owner.Update(message.SequenceId, 0, message.Content.Fallback);
|
|
target.Update(target.LastReadSequenceId, target.UnreadCount + 1, message.Content.Fallback);
|
|
|
|
messageDb.Conversations.UpdateRange(owner,target);
|
|
await messageDb.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
await endpoint.Publish(message.ToIntegrationEvent());
|
|
}
|
|
|
|
public async Task Handle(MessageWithdrawDomainEvent notification, CancellationToken cancellationToken)
|
|
{
|
|
var message = notification.Message;
|
|
await endpoint.Publish(new MsgWithdrawEvent(message.Id, message.State.ToString(), message.StreamKey), cancellationToken);
|
|
}
|
|
}
|
|
}
|