XCKJ
/
irc-netcore-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改中心调研表逻辑
continuous-integration/drone/push Build is passing Details

IRC_NewDev
hang 2024-10-18 15:08:28 +08:00
parent a7277be2ba
commit 2a0d01c161
5 changed files with 60 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
IRaCIS.Core.API/HostService/HangfireHostService.cs
View File

@ -14,11 +14,13 @@ using MassTransit.Scheduling;
using Hangfire.Storage;
using System.Linq;
using Microsoft.EntityFrameworkCore;
using MassTransit.Mediator;
namespace IRaCIS.Core.API.HostService;
public class HangfireHostService(IRecurringMessageScheduler _recurringMessageScheduler,
IRepository<TrialEmailNoticeConfig> _trialEmailNoticeConfigRepository,
IMediator _mediator,
ILogger<HangfireHostService> _logger) : IHostedService
{
@ -53,7 +55,7 @@ public class HangfireHostService(IRecurringMessageScheduler _recurringMessageSch
}
await _recurringMessageScheduler.ScheduleRecurringPublish(new QCImageQuestionSchedule() { CronExpression = "0/3 * * * * ? " }, new MasstransiTestCommand { value = "message at " + DateTime.Now.ToString() });
//await _recurringMessageScheduler.ScheduleRecurringPublish(new QCImageQuestionSchedule() { CronExpression = "0/3 * * * * ? " }, new MasstransiTestCommand { value = "message at " + DateTime.Now.ToString() });

3
IRaCIS.Core.API/_ServiceExtensions/MassTransitSetup.cs
View File

@ -39,12 +39,15 @@ namespace IRaCIS.Core.API
cfg.UsePublishMessageScheduler();
cfg.UseConsumeFilter(typeof(ConsumeExceptionFilter<>), context,
x => x.Include(type => type.IsAssignableTo(typeof(DomainEvent))));
cfg.UseConsumeFilter(typeof(CultureInfoFilter<>), context,
x => x.Include(type => type.IsAssignableTo(typeof(DomainEvent))));
cfg.ConfigureEndpoints(context); // 自动配置所有消费者的端点
});
});

25
IRaCIS.Core.Application/MassTransit/Consumer/SiteSurverEmailConsumer.cs
View File

@ -219,26 +219,37 @@ public class SiteSurverRejectedEventConsumer(
var trialId = siteSurveyInfo.TrialId;
var messageToSend = new MimeMessage();
var name = siteSurveyInfo.UserName;
var toUserName = siteSurveyInfo.UserName;
if (context.Message.IsHaveSPMOrCPM)
{
//PM 驳回到SPM
if (siteSurveyInfo.State == TrialSiteSurveyEnum.CRCSubmitted)
{
var user = await _userRepository.FirstOrDefaultAsync(t => t.Id == siteSurveyInfo.PreliminaryUserId);
//var user = await _userRepository.FirstOrDefaultAsync(t => t.Id == siteSurveyInfo.PreliminaryUserId);
name = user.FullName;
//name = user.FullName;
var sPMOrCPMList = _trialRepository.Where(t => t.Id == trialId)
.SelectMany(t => t.TrialUserList)
.Where(t => t.User.UserTypeEnum == UserTypeEnum.SPM || t.User.UserTypeEnum == UserTypeEnum.CPM)
.Select(t => new { t.User.EMail, t.User.FullName, t.User.UserTypeEnum }).ToList();
foreach (var user in sPMOrCPMList)
{
messageToSend.To.Add(new MailboxAddress(user.FullName, user.EMail));
}
toUserName = string.Join('、', sPMOrCPMList.Select(t => t.FullName));
messageToSend.To.Add(new MailboxAddress(name, user.EMail));
}
//SPM 驳回到CRC
else if (siteSurveyInfo.State == TrialSiteSurveyEnum.ToSubmit)
{
messageToSend.To.Add(new MailboxAddress(name, siteSurveyInfo.Email));
messageToSend.To.Add(new MailboxAddress(toUserName, siteSurveyInfo.Email));
}
}
@ -265,7 +276,7 @@ public class SiteSurverRejectedEventConsumer(
{
var topicStr = string.Format(input.topicStr, companyName, trialInfo.ResearchProgramNo);
var htmlBodyStr = string.Format(CommonEmailHelper.ReplaceCompanyName(_systemEmailConfig, input.htmlBodyStr),
name,
toUserName,
trialInfo.TrialCode,
trialInfo.ResearchProgramNo,
trialInfo.ExperimentName,

33
IRaCIS.Core.Application/MassTransit/Extension/ConsumeExceptionFilter.cs Normal file
View File

@ -0,0 +1,33 @@
using System;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using IRaCIS.Core.Domain.BaseModel;
using IRaCIS.Core.Domain.Share;
using MassTransit;
using Microsoft.Extensions.Logging;
using NPOI.SS.Formula.Functions;
public class ConsumeExceptionFilter<T>(ILogger<ConsumeExceptionFilter<T>> _logger) : IFilter<ConsumeContext<T>> where T : DomainEvent
{
public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
{
try
{
await next.Send(context);
}
catch (Exception exception)
{
var errorInfo = $"Exception: {exception.Message}[{exception.StackTrace}]" + (exception.InnerException != null ? $" InnerException: {exception.InnerException.Message}[{exception.InnerException.StackTrace}]" : "");
_logger.LogError(errorInfo);
}
}
public void Probe(ProbeContext context)
{
context.CreateFilterScope("ConsumeException");
}
}

4
IRaCIS.Core.Infra.EFCore/Interceptor/DispatchDomainEventsInterceptor.cs
View File

@ -14,7 +14,7 @@ namespace IRaCIS.Core.Infra.EFCore.Interceptor
// 发件箱模式参考： https://dev.to/antonmartyniuk/use-masstransit-to-implement-outbox-pattern-with-ef-core-and-mongodb-oep#:~:text=MongoDB%20replica%20set%20is%20required%20for%20both%20publisher%20and%20consumer
// 1、IPublishEndpoint 才会将事件存储到发件箱表中， 高级IBus接口时 - 消息不会存储在发件箱中，必须有savechanges 才会一起提交保存到数据库中
// 2、进入消息代理之前，发布事件在OutboxState OutboxMessage， 进入消费者以后（已经删除OutboxState OutboxMessage），消费失败，需要修改代码重新发布，然后之前消费事件的重新处理，错误处理参考：https://www.youtube.com/watch?v=3TMKUu7c4lc
public class DispatchDomainEventsInterceptor(IMediator _mediator, IMessageScheduler _scheduler, IPublishEndpoint _publishEndpoint) : SaveChangesInterceptor
public class DispatchDomainEventsInterceptor(IMediator _mediator, IMessageScheduler _scheduler/*, IPublishEndpoint _publishEndpoint*/) : SaveChangesInterceptor
{
//领域事件通常与数据变更密切相关。如果在 SaveChanges 之前发布事件，有可能事件发布时的数据状态还没有被持久化到数据库。这可能导致事件消费者看到的是一个不一致的状态
@ -64,7 +64,7 @@ namespace IRaCIS.Core.Infra.EFCore.Interceptor
await _scheduler.SchedulePublish(DateTime.Now.AddSeconds((int)domainEvent.DelaySeconds!), (object)domainEvent);
}
await _publishEndpoint.Publish(domainEvent.GetType(), domainEvent);
await _mediator.Publish(domainEvent.GetType(), domainEvent);
}
}