diff --git a/Directory.Build.props b/Directory.Build.props index 7d6f04d0..1f5eaf03 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,6 +1,9 @@ + + - 8.0.5.17 + + 8.0.6 diff --git a/Directory.Build.targets b/Directory.Build.targets new file mode 100644 index 00000000..34e847f5 --- /dev/null +++ b/Directory.Build.targets @@ -0,0 +1,26 @@ + + + + + <_LocalFakeCandidate Include="@(PackageReference -> '$(FakeRepoRoot)\src\%(Identity)\%(Identity).csproj')" /> + <_LocalFakeCandidate Include="@(PackageReference -> '$(FakeRepoRoot)\modules\rbac\%(Identity)\%(Identity).csproj')" /> + <_LocalFakeCandidate Include="@(PackageReference -> '$(FakeRepoRoot)\modules\file-management\%(Identity)\%(Identity).csproj')" /> + + + + + + + + + + <_MissingLocalFake Include="@(_LocalFakeCandidate)" Condition="!Exists('%(Identity)')" /> + + + + diff --git a/Fake.Local.props b/Fake.Local.props new file mode 100644 index 00000000..8519e293 --- /dev/null +++ b/Fake.Local.props @@ -0,0 +1,10 @@ + + + + C:\Users\BSI\RiderProjects\fake2 + false + + diff --git a/TrackCenter.sln b/TrackCenter.sln index 3bb0c534..06261b70 100644 --- a/TrackCenter.sln +++ b/TrackCenter.sln @@ -20,6 +20,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "solution items", "solution .editorconfig = .editorconfig Directory.Build.props = Directory.Build.props .gitea\workflows\deploy.yml = .gitea\workflows\deploy.yml + Fake.Local.props = Fake.Local.props + Directory.Build.targets = Directory.Build.targets EndProjectSection EndProject Global diff --git a/src/TrackCenter.Api/TrackCenter.Api.csproj b/src/TrackCenter.Api/TrackCenter.Api.csproj index 7022770f..48e85830 100644 --- a/src/TrackCenter.Api/TrackCenter.Api.csproj +++ b/src/TrackCenter.Api/TrackCenter.Api.csproj @@ -1,4 +1,4 @@ - + @@ -20,6 +20,9 @@ + + + diff --git a/src/TrackCenter.Api/TrackCenterApiModule.cs b/src/TrackCenter.Api/TrackCenterApiModule.cs index 4681cf88..7af17e9d 100644 --- a/src/TrackCenter.Api/TrackCenterApiModule.cs +++ b/src/TrackCenter.Api/TrackCenterApiModule.cs @@ -5,6 +5,8 @@ using Fake.AspNetCore.ExceptionHandling; using Fake.AspNetCore.Newtonsoft; using Fake.AspNetCore.Swagger; using Fake.Autofac; +using Fake.EventBus.Dashboard; +using Fake.EventBus.IntegrationEventLog; using Fake.FeiShu; using Fake.Json; using Fake.Modularity; @@ -15,6 +17,7 @@ using Hangfire; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.HttpOverrides; using Microsoft.AspNetCore.Mvc.Authorization; +using Microsoft.EntityFrameworkCore; using OpenTelemetry.Resources; using OpenTelemetry.Trace; using TrackCenter.Application; @@ -30,6 +33,9 @@ namespace TrackCenter.Api; [DependsOn(typeof(FakeAspNetCoreModule))] [DependsOn(typeof(TrackCenterApplicationModule))] [DependsOn(typeof(TrackCenterRepositoryModule))] +// RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox（与 admin 一致） +[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))] +[DependsOn(typeof(FakeEventBusDashboardModule))] [DependsOn(typeof(FakeSerilogModule))] public class TrackCenterApiModule : FakeModule { @@ -53,6 +59,20 @@ public class TrackCenterApiModule : FakeModule // 配置签名验证选项 services.Configure(configuration.GetSection("SignatureValidation")); + // Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务 + services.AddDbContext(options => + { + var connectionString = configuration.GetConnectionString("IntegrationEventLog") + ?? configuration.GetConnectionString("Default") + ?? configuration["ConnectionStrings:Default"] + ?? string.Empty; + var serverVersion = ServerVersion.AutoDetect(connectionString); + options.UseMySql(connectionString, serverVersion); +#if DEBUG + options.EnableSensitiveDataLogging(); +#endif + }); + // 全局授权配置：所有接口默认需要授权 services.AddMvc(options => { @@ -186,6 +206,9 @@ public class TrackCenterApiModule : FakeModule Authorization = [new AnonymousHangfireAuthorizationFilter()] }); + // 必须在 UsePathBase 之后：完整对外路径为 {PathBase}/eventbus + app.UseFakeEventBusDashboard(); + app.Services.GetRequiredService().Send("Application started"); } } diff --git a/src/TrackCenter.Api/appsettings.json b/src/TrackCenter.Api/appsettings.json index 06ba18d7..c004a63e 100644 --- a/src/TrackCenter.Api/appsettings.json +++ b/src/TrackCenter.Api/appsettings.json @@ -3,5 +3,8 @@ "Consul": { "Address": "http://test.ideship.com:30850", "Token": "c0aea902-dca0-ac29-299b-a0d47e5402aa" + }, + "DataSeed": { + "Enable": false } } diff --git a/src/TrackCenter.Application/CAP/CAPFilter.cs b/src/TrackCenter.Application/CAP/CAPFilter.cs deleted file mode 100644 index 9e79b526..00000000 --- a/src/TrackCenter.Application/CAP/CAPFilter.cs +++ /dev/null @@ -1,69 +0,0 @@ -using DotNetCore.CAP.Filter; -using InterfaceForward.Domain.Shared.Helpers; -using TrackCenter.Application.Helpers; - -namespace TrackCenter.Application.CAP; - -public class CapFilter : SubscribeFilter -{ - /// - /// 执行前 - /// - /// - /// - public override async Task OnSubscribeExecutingAsync(ExecutingContext context) - { - // 轨迹新增、更新要做幂等 - if (!context.DeliverMessage.Headers["cap-msg-name"] - .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe)) return; - var argumentJsonStr = context.Arguments.ToJson(); - - if (!string.IsNullOrEmpty(argumentJsonStr)) - { - context.DeliverMessage.Headers.TryGetValue("track-arguments-hash", out var hash); - if (hash.IsNullOrEmpty()) // 计算参数hash - { - hash = StringHelper.GetHash(argumentJsonStr); - context.DeliverMessage.Headers["track-arguments-hash"] = hash; - } - - if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}{DateTime.Now:G}")) - { - throw new CapNoRetryException($"检测到重复消费,消费参数：{argumentJsonStr}"); - } - } - } - - public override async Task OnSubscribeExecutedAsync(ExecutedContext context) - { - context.DeliverMessage.Headers.TryGetValue("track-arguments-hash", out var hash); - if (!hash.IsNullOrEmpty()) - { - await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}{DateTime.Now:G}", 1, TimeSpan.FromMinutes(3)); - } - } - - /// - /// 消费异常事件 - /// - /// - /// - public override Task OnSubscribeExceptionAsync(ExceptionContext context) - { - context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); - - if (context.Exception is CapNoRetryException ex) - { - context.ExceptionHandled = true; - Log.Warning(ex.ToString(), true); - } - else - { - // 发飞书，异常继续上抛，进入重试机制 - // tips：重试有冷却期（默认4分钟） - Log.Error($"cap-msg-id：{messageId}\n{context.Exception}"); - } - - return Task.CompletedTask; - } -} diff --git a/src/TrackCenter.Application/CAP/CapNoRetryException.cs b/src/TrackCenter.Application/CAP/CapNoRetryException.cs deleted file mode 100644 index 4047a271..00000000 --- a/src/TrackCenter.Application/CAP/CapNoRetryException.cs +++ /dev/null @@ -1,12 +0,0 @@ -using Fake; -using Fake.ExceptionHandling; -using Fake.Logging; -using Microsoft.Extensions.Logging; - -namespace TrackCenter.Application.CAP; - -public class CapNoRetryException(string? message = null, Exception? innerException = null) - : FakeException(message, innerException), IHasLogLevel -{ - public LogLevel LogLevel { get; set; } = LogLevel.Warning; -} diff --git a/src/TrackCenter.Application/CAP/LoggerExtensions.cs b/src/TrackCenter.Application/CAP/LoggerExtensions.cs deleted file mode 100644 index aadf7184..00000000 --- a/src/TrackCenter.Application/CAP/LoggerExtensions.cs +++ /dev/null @@ -1,90 +0,0 @@ -using Microsoft.Extensions.Logging; -using ILogger = Microsoft.Extensions.Logging.ILogger; - -namespace TrackCenter.Application.CAP; - -internal static class LoggerExtensions -{ - public static void ConsumerExecutedAfterThreshold(this ILogger logger, string messageId, int retries) - { - logger.LogWarning( - $"The Subscriber of the message({messageId}) still fails after {retries}th executions and we will stop retrying."); - } - - public static void SenderAfterThreshold(this ILogger logger, string messageId, int retries) - { - logger.LogWarning( - $"The Publisher of the message({messageId}) still fails after {retries}th sends and we will stop retrying."); - } - - public static void ExecutedThresholdCallbackFailed(this ILogger logger, Exception ex) - { - logger.LogWarning(ex, "FailedThresholdCallback action raised an exception:" + ex.Message); - } - - public static void ConsumerDuplicates(this ILogger logger, string subscriber, string group) - { - logger.LogWarning( - $"We detected that you have duplicate subscribers ({subscriber}) in same group ({group}), this will cause diversity behavior."); - } - - public static void ConsumerExecutionRetrying(this ILogger logger, string messageId, int retries) - { - logger.LogWarning($"The {retries}th retrying consume a message failed. message id: {messageId}"); - } - - public static void SenderRetrying(this ILogger logger, string messageId, int retries) - { - logger.LogWarning($"The {retries}th retrying send a message failed. message id: {messageId} "); - } - - public static void MessageReceived(this ILogger logger, string messageId, string name) - { - logger.LogDebug($"Received message. id:{messageId}, name: {name}"); - } - - public static void MessagePublishException(this ILogger logger, string? messageId, string reason, Exception? ex) - { - logger.LogError(ex, - $"An exception occurred while publishing a message, reason:{reason}. message id:{messageId}"); - } - - public static void ConsumerExecuting(this ILogger logger, string className, string methodName, string group) - { - logger.LogInformation($"Executing subscriber method '{className}.{methodName}' on group '{group}'"); - } - - public static void ConsumerExecuted(this ILogger logger, string className, string methodName, string group, - double milliseconds, string? instance) - { - logger.LogInformation( - $"Executed subscriber method '{className}.{methodName}' on group '{group}' with instance '{instance}' in {milliseconds}ms"); - } - - public static void ConsumerExecuteFailed(this ILogger logger, string topic, string id, string? instance, - Exception? ex) - { - logger.LogError(ex, - $"An exception occurred while executing the subscription method. Topic:{topic}, Id:{id}, Instance: {instance}"); - } - - public static void ServerStarting(this ILogger logger) - { - logger.LogInformation("Starting the processing server."); - } - - public static void ProcessorsStartedError(this ILogger logger, Exception ex) - { - logger.LogError(ex, "Starting the processors throw an exception."); - } - - public static void ServerShuttingDown(this ILogger logger) - { - logger.LogInformation("Shutting down the processing server..."); - } - - public static void ExpectedOperationCanceledException(this ILogger logger, Exception ex) - { - logger.LogWarning(ex, $"Expected an OperationCanceledException, but found '{ex.Message}'."); - } -} diff --git a/src/TrackCenter.Application/CAP/MySubscribeExecutor.cs b/src/TrackCenter.Application/CAP/MySubscribeExecutor.cs deleted file mode 100644 index a70f1408..00000000 --- a/src/TrackCenter.Application/CAP/MySubscribeExecutor.cs +++ /dev/null @@ -1,269 +0,0 @@ -using System.Diagnostics; -using System.Reflection; -using DotNetCore.CAP; -using DotNetCore.CAP.Diagnostics; -using DotNetCore.CAP.Internal; -using DotNetCore.CAP.Messages; -using DotNetCore.CAP.Persistence; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using ILogger = Microsoft.Extensions.Logging.ILogger; - -namespace TrackCenter.Application.CAP; - -public class MySubscribeExecutor : ISubscribeExecutor -{ - // diagnostics listener - // ReSharper disable once InconsistentNaming - private static readonly DiagnosticListener s_diagnosticListener = - new(CapDiagnosticListenerNames.DiagnosticListenerName); - - private readonly IDataStorage _dataStorage; - private readonly string? _hostName; - private readonly ILogger _logger; - private readonly CapOptions _options; - private readonly IServiceProvider _provider; - - public MySubscribeExecutor( - ILogger logger, - IOptions options, - IServiceProvider provider) - { - _provider = provider; - _logger = logger; - _options = options.Value; - - _dataStorage = _provider.GetRequiredService(); - Invoker = _provider.GetRequiredService(); - _hostName = Helper.GetInstanceHostname(); - } - - private ISubscribeInvoker Invoker { get; } - - public async Task ExecuteAsync(MediumMessage message, ConsumerExecutorDescriptor? descriptor = null, - CancellationToken cancellationToken = default) - { - if (descriptor == null) - { - var selector = _provider.GetRequiredService(); - if (!selector.TryGetTopicExecutor(message.Origin.GetName(), message.Origin.GetGroup()!, out descriptor)) - { - var error = - $"Message (Name:{message.Origin.GetName()},Group:{message.Origin.GetGroup()}) can not be found subscriber." + - $"{Environment.NewLine} see: https://github.com/dotnetcore/CAP/issues/63"; - _logger.LogError(error); - - TracingError(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), message.Origin, null, - new Exception(error)); - - var ex = new SubscriberNotFoundException(error); - await SetFailedState(message, ex); - return OperateResult.Failed(ex); - } - } - - bool retry; - OperateResult result; - - //record instance id - message.Origin.Headers[Headers.ExecutionInstanceId] = _hostName; - - do - { - var (shouldRetry, operateResult) = await ExecuteWithoutRetryAsync(message, descriptor, cancellationToken) - .ConfigureAwait(false); - result = operateResult; - if (result.Equals(OperateResult.Success)) return result; - retry = shouldRetry; - } while (retry); - - return result; - } - - private async Task<(bool, OperateResult)> ExecuteWithoutRetryAsync(MediumMessage message, - ConsumerExecutorDescriptor descriptor, CancellationToken cancellationToken) - { - if (message == null) throw new ArgumentNullException(nameof(message)); - - cancellationToken.ThrowIfCancellationRequested(); - - try - { - _logger.ConsumerExecuting(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name, - descriptor.Attribute.Group); - - var sp = Stopwatch.StartNew(); - - await InvokeConsumerMethodAsync(message, descriptor, cancellationToken).ConfigureAwait(false); - - sp.Stop(); - - await SetSuccessfulState(message).ConfigureAwait(false); - - CapEventCounterSource.Log.WriteInvokeTimeMetrics(sp.Elapsed.TotalMilliseconds); - _logger.ConsumerExecuted(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name, - descriptor.Attribute.Group, sp.Elapsed.TotalMilliseconds, message.Origin.GetExecutionInstanceId()); - - return (false, OperateResult.Success); - } - catch (Exception ex) - { - _logger.ConsumerExecuteFailed(message.Origin.GetName(), message.DbId, - message.Origin.GetExecutionInstanceId(), ex); - - return (await SetFailedState(message, ex).ConfigureAwait(false), OperateResult.Failed(ex)); - } - } - - private Task SetSuccessfulState(MediumMessage message) - { - message.ExpiresAt = DateTime.Now.AddSeconds(_options.SucceedMessageExpiredAfter); - - return _dataStorage.ChangeReceiveStateAsync(message, StatusName.Succeeded); - } - - private async Task SetFailedState(MediumMessage message, Exception ex) - { - if (ex is SubscriberNotFoundException) - message.Retries = _options.FailedRetryCount; // not retry if SubscriberNotFoundException - - var needRetry = UpdateMessageForRetry(message); - - message.Origin.AddOrUpdateException(ex); - message.ExpiresAt = message.Added.AddSeconds(_options.FailedMessageExpiredAfter); - - await _dataStorage.ChangeReceiveStateAsync(message, StatusName.Failed).ConfigureAwait(false); - - return needRetry; - } - - private bool UpdateMessageForRetry(MediumMessage message) - { - var retries = ++message.Retries; - - if (retries == _options.FailedRetryCount) - { - try - { - _options.FailedThresholdCallback?.Invoke(new FailedInfo - { - ServiceProvider = _provider, - MessageType = MessageType.Subscribe, - Message = message.Origin - }); - - _logger.ConsumerExecutedAfterThreshold(message.DbId, _options.FailedRetryCount); - } - catch (Exception ex) - { - _logger.ExecutedThresholdCallbackFailed(ex); - } - } - - return false; - } - - private async Task InvokeConsumerMethodAsync(MediumMessage message, ConsumerExecutorDescriptor descriptor, - CancellationToken cancellationToken) - { - var consumerContext = new ConsumerContext(descriptor, message); - var tracingTimestamp = TracingBefore(message.Origin, descriptor.MethodInfo); - try - { - var ret = await Invoker.InvokeAsync(consumerContext, cancellationToken).ConfigureAwait(false); - - TracingAfter(tracingTimestamp, message.Origin, descriptor.MethodInfo); - - if (!string.IsNullOrEmpty(ret.CallbackName)) - { - var header = new Dictionary - { - [Headers.CorrelationId] = message.Origin.GetId(), - [Headers.CorrelationSequence] = (message.Origin.GetCorrelationSequence() + 1).ToString() - }; - - if (message.Origin.Headers.TryGetValue(Headers.TraceParent, out var traceparent)) - header[Headers.TraceParent] = traceparent; - - await _provider.GetRequiredService() - .PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) - { - //ignore - } - catch (Exception ex) - { - var e = new SubscriberExecutionFailedException(ex.Message, ex); - - TracingError(tracingTimestamp, message.Origin, descriptor.MethodInfo, e); - - e.ReThrow(); - } - } - - #region tracing - - private static long? TracingBefore(Message message, MethodInfo method) - { - if (s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.BeforeSubscriberInvoke)) - { - var eventData = new CapEventDataSubExecute - { - OperationTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), - Operation = message.GetName(), - Message = message, - MethodInfo = method - }; - - s_diagnosticListener.Write(CapDiagnosticListenerNames.BeforeSubscriberInvoke, eventData); - - return eventData.OperationTimestamp; - } - - return null; - } - - private static void TracingAfter(long? tracingTimestamp, Message message, MethodInfo method) - { - CapEventCounterSource.Log.WriteInvokeMetrics(); - if (tracingTimestamp != null && - s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.AfterSubscriberInvoke)) - { - var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); - var eventData = new CapEventDataSubExecute - { - OperationTimestamp = now, - Operation = message.GetName(), - Message = message, - MethodInfo = method, - ElapsedTimeMs = now - tracingTimestamp.Value - }; - - s_diagnosticListener.Write(CapDiagnosticListenerNames.AfterSubscriberInvoke, eventData); - } - } - - private static void TracingError(long? tracingTimestamp, Message message, MethodInfo? method, Exception ex) - { - if (tracingTimestamp != null && - s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.ErrorSubscriberInvoke)) - { - var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); - var eventData = new CapEventDataSubExecute - { - OperationTimestamp = now, - Operation = message.GetName(), - Message = message, - MethodInfo = method, - ElapsedTimeMs = now - tracingTimestamp.Value, - Exception = ex - }; - - s_diagnosticListener.Write(CapDiagnosticListenerNames.ErrorSubscriberInvoke, eventData); - } - } - - #endregion -} diff --git a/src/TrackCenter.Application/CAP/SubscriberExecutionFailedException.cs b/src/TrackCenter.Application/CAP/SubscriberExecutionFailedException.cs deleted file mode 100644 index a3616edc..00000000 --- a/src/TrackCenter.Application/CAP/SubscriberExecutionFailedException.cs +++ /dev/null @@ -1,15 +0,0 @@ -namespace TrackCenter.Application.CAP; - -internal sealed class SubscriberExecutionFailedException : Exception -{ - private readonly Exception _originException; - - public SubscriberExecutionFailedException(string message, Exception ex) : base(message, ex) - { - _originException = ex; - } - - public override string? StackTrace => _originException.StackTrace; - - public override string? Source => _originException.Source; -} diff --git a/src/TrackCenter.Application/Events/OrderPodPullEvent.cs b/src/TrackCenter.Application/Events/OrderPodPullEvent.cs index 3cfcf71c..3d09e256 100644 --- a/src/TrackCenter.Application/Events/OrderPodPullEvent.cs +++ b/src/TrackCenter.Application/Events/OrderPodPullEvent.cs @@ -1,6 +1,12 @@ +using Fake.EventBus.Distributed; + namespace TrackCenter.Application.Events; -public class OrderPodPullEvent +/// +/// 订单 POD 拉取（当前仓库无订阅方，仍走 Outbox 投递以兼容后续消费者）。 +/// +[Serializable] +public class OrderPodPullEvent : IntegrationEvent { // 确定应用 public string AppKey { get; set; } diff --git a/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs b/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs index 5259578d..e438598a 100644 --- a/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs +++ b/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs @@ -1,6 +1,9 @@ +using Fake.EventBus.Distributed; + namespace TrackCenter.Application.Events; -public class TrackChangeAfterUpdateOrderEvent +[Serializable] +public class TrackChangeAfterUpdateOrderEvent : IntegrationEvent { /// /// 订单Id diff --git a/src/TrackCenter.Application/Events/TrackDeleteEvent.cs b/src/TrackCenter.Application/Events/TrackDeleteEvent.cs index bb666875..8cac8130 100644 --- a/src/TrackCenter.Application/Events/TrackDeleteEvent.cs +++ b/src/TrackCenter.Application/Events/TrackDeleteEvent.cs @@ -1,8 +1,10 @@ +using Fake.EventBus.Distributed; using TrackCenter.Application.Dtos.ProvideServices; namespace TrackCenter.Application.Events; -public class TrackDeleteEvent +[Serializable] +public class TrackDeleteEvent : IntegrationEvent { /// /// 根据esId定位轨迹，优先级最高 diff --git a/src/TrackCenter.Application/Events/TrackPushEvent.cs b/src/TrackCenter.Application/Events/TrackPushEvent.cs index 001c35b4..53a63a8e 100644 --- a/src/TrackCenter.Application/Events/TrackPushEvent.cs +++ b/src/TrackCenter.Application/Events/TrackPushEvent.cs @@ -1,10 +1,13 @@ +using Fake.EventBus.Distributed; + namespace TrackCenter.Application.Events; /// /// 轨迹推送事件 /// 每次可传递多个订单，但仅能传递同一个轨迹内容 /// -public class TrackPushEvent +[Serializable] +public class TrackPushEvent : IntegrationEvent { /// /// 操作类型 diff --git a/src/TrackCenter.Application/Events/TrackUpsertEvent.cs b/src/TrackCenter.Application/Events/TrackUpsertEvent.cs index 503161c6..ee4ac79f 100644 --- a/src/TrackCenter.Application/Events/TrackUpsertEvent.cs +++ b/src/TrackCenter.Application/Events/TrackUpsertEvent.cs @@ -1,8 +1,10 @@ +using Fake.EventBus.Distributed; using TrackCenter.Application.Dtos.ProvideServices; namespace TrackCenter.Application.Events; -public class TrackUpsertEvent +[Serializable] +public class TrackUpsertEvent : IntegrationEvent { /// /// 应用 diff --git a/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs b/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs new file mode 100644 index 00000000..ef946ab7 --- /dev/null +++ b/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs @@ -0,0 +1,9 @@ +namespace TrackCenter.Application.Events; + +/// +/// 轨迹刷新引发的新增/更新（独立队列，与 隔离）。 +/// +[Serializable] +public class TrackUpsertForRefreshEvent : TrackUpsertEvent +{ +} diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs similarity index 93% rename from src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs index 0d801fbf..0da20f58 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs @@ -1,17 +1,18 @@ -using DotNetCore.CAP; +using Fake.EventBus; +using Fake.EventBus.Distributed; using Fake.UnitOfWork; using TrackCenter.Application.Events; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Domain.Aggregates.OrderAggregate; -namespace TrackCenter.Application.CAP.EventHandlers; +namespace TrackCenter.Application.Messaging; public class TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository, IServiceProvider serviceProvider) - : ITransientDependency, ICapSubscribe + : IEventHandler, ITransientDependency { - [UnitOfWork] - [CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)] - public virtual async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event) + public virtual async Task HandleAsync( + TrackChangeAfterUpdateOrderEvent @event, + CancellationToken cancellationToken = default) { var orders = await GetOrdersAsync(@event.OrderIds); diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs similarity index 80% rename from src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs index c3531b83..09105b19 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs @@ -1,4 +1,5 @@ -using DotNetCore.CAP; +using Fake.EventBus; +using Fake.EventBus.Distributed; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.ProvideServices; @@ -11,40 +12,20 @@ using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Repository.Aggregates.OrderAggregate; -namespace TrackCenter.Application.CAP.EventHandlers; +namespace TrackCenter.Application.Messaging; -public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe +public class TrackDeleteEventHandler( + ProvideQueries provideQueries, + IOrderRepository orderRepository, + IDistributedEventBus eventBus, + ITrackDetailsEsRepository trackDetailsEsRepository, + IOrderOperationRecordEsRepository orderOperationRecordEsRepository, + BroadcastingService broadcastingService, + WarningSignalPublisher warningSignalPublisher) + : IEventHandler, ITransientDependency { - private readonly ICapPublisher _capPublisher; - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - private readonly IOrderRepository _orderRepository; - private readonly ProvideQueries _provideQueries; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly BroadcastingService _broadcastingService; - private readonly WarningSignalPublisher _warningSignalPublisher; - - public TrackDeleteEventHandler( - ProvideQueries provideQueries, - IOrderRepository orderRepository, - ICapPublisher capPublisher, - ITrackDetailsEsRepository trackDetailsEsRepository, - IOrderOperationRecordEsRepository orderOperationRecordEsRepository, - BroadcastingService broadcastingService, - WarningSignalPublisher warningSignalPublisher - ) - { - _provideQueries = provideQueries; - _orderRepository = orderRepository; - _capPublisher = capPublisher; - _trackDetailsEsRepository = trackDetailsEsRepository; - _orderOperationRecordEsRepository = orderOperationRecordEsRepository; - _broadcastingService = broadcastingService; - _warningSignalPublisher = warningSignalPublisher; - } - - [UnitOfWork] - [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] - public virtual async Task ConsumerAsync(TrackDeleteEvent @event) + [UnitOfWork(IsTransactional = true)] + public virtual async Task HandleAsync(TrackDeleteEvent @event, CancellationToken cancellationToken = default) { if (!CanDelete(@event)) return; @@ -57,7 +38,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private async Task DownByOrderLink(TrackDeleteEvent @event) { - var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); + var linkedOrders = await orderRepository.GetLinkedOrdersAsync(@event.OrderIds); foreach (var orderId in @event.OrderIds) { foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.AppKey)) @@ -70,7 +51,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe // 不同应用需要做内容映射 if (@event.AppKey != nextEvent.AppKey) { - var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( + var codeMaps = await provideQueries.GetTrackCodeMapsAsync( @event.AppKey, nextEvent.AppKey, @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() @@ -95,7 +76,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe { var subEvent = nextEvent.DeepCopy(); subEvent.OrderIds = subOrderIds; - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); + // 关联单继续走删除链路（原 CAP 误发到 Upsert topic，此处修正） + await eventBus.PublishAsync(subEvent); } } } @@ -125,7 +107,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe // 优先根据EsId查询轨迹 if (@event.EsIds.Length > 0) { - deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); + deleteTracks = await trackDetailsEsRepository.GetDataByIds(@event.EsIds); // 外部api获取的轨迹不允许删除 if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return; @@ -143,12 +125,12 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe }).ToList(); // 查出订单下的所有轨迹，以供后面回退使用 - oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + oldTracks = await trackDetailsEsRepository.GetListAsync(@event.OrderIds); } else { // 根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新 - oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + oldTracks = await trackDetailsEsRepository.GetListAsync(@event.OrderIds); // 构建轨迹标识符 var trackIds = @event.TrackIdentifiers.Select(y => $"{y.TrackCode}{y.GetUtc0TrackTime():G}").ToHashSet(); @@ -177,7 +159,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe if (deleteTracks.Count == 0) return; // 删除轨迹批量 - await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); + await trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); // 删除轨迹后，更新订单上的轨迹信息 await TrackDeleteAfterUpdateOrderAsync(@event, oldTracks, deleteTracks, orderIds); @@ -186,13 +168,13 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe await PublicTrackTopicAsync(@event, deleteTracks); // 发布预警信号，稍后将进行预警计算 - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, deleteTracks); + warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, deleteTracks); } private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List delTracks) { var trackContentIds = delTracks.Select(x => x.TrackContentId).ToHashSet(); - var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds); + var trackContents = await provideQueries.GetTrackContentAsync(trackContentIds); var contentDict = trackContents.ToDictionary(x => x.Id); var needPublicTracks = delTracks @@ -255,7 +237,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe }; var content = contentDict[track.TrackContentId]; - _broadcastingService.TrackChangePublicTopic(eto, content.AppKey); + broadcastingService.TrackChangePublicTopic(eto, content.AppKey); } } @@ -270,7 +252,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); var newTracks = oldTracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); - var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), + var orders = await orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, @@ -315,7 +297,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe } //更新订单上冗余轨迹明细的字段 - await _orderRepository.UpdateRangeAsync(orders, + await orderRepository.UpdateRangeAsync(orders, x => new { x.LatestTrackStatus, @@ -329,7 +311,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe }); //批量写入订单操作日志 - _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex + _ = orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex { OrderId = order.Id, OrderNo = order.OrderNo, diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs similarity index 91% rename from src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs index c9c8f06c..206796f1 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using System.Text.Json; -using DotNetCore.CAP; +using Fake.EventBus; +using Fake.EventBus.Distributed; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Options; @@ -18,48 +19,28 @@ using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Aggregates.TrackPushAggregate; using TrackCenter.Repository.Otel; -namespace TrackCenter.Application.CAP.EventHandlers; +namespace TrackCenter.Application.Messaging; -public class TrackPushEventHandler : ITransientDependency, ICapSubscribe +public class TrackPushEventHandler( + IOrderRepository orderRepository, + TrackPushCacheService trackPushCacheService, + ITrackPushRecordEsRepository trackPushRecordEsRepository, + IOptionsMonitor options, + BroadcastingService broadcastingService, + ITrackDetailsEsRepository trackDetailsEsRepository, + InterfaceForwardService interfaceForwardService) + : IEventHandler, ITransientDependency { - private readonly IOptionsMonitor _options; - private readonly IOrderRepository _orderRepository; - private readonly TrackPushCacheService _trackPushCacheService; - private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly InterfaceForwardService _interfaceForwardService; - private readonly BroadcastingService _broadcastingService; private List _orders = null!; private List _pushConfigs = null!; private Dictionary> _orderTypeSortedConfigsDict = null!; //订单类型,根据sort从小到大排序后的配置字典 - public TrackPushEventHandler( - IOrderRepository orderRepository, - TrackPushCacheService trackPushCacheService, - ITrackPushRecordEsRepository trackPushRecordEsRepository, - IOptionsMonitor options, - BroadcastingService broadcastingService, - ITrackDetailsEsRepository trackDetailsEsRepository, - InterfaceForwardService interfaceForwardService) - { - _orderRepository = orderRepository; - _trackPushCacheService = trackPushCacheService; - _trackPushRecordEsRepository = trackPushRecordEsRepository; - _options = options; - _broadcastingService = broadcastingService; - _trackDetailsEsRepository = trackDetailsEsRepository; - _interfaceForwardService = interfaceForwardService; - } - - [UnitOfWork] - [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] - [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] - public virtual async Task ConsumerAsync(TrackPushEvent @event) + public virtual async Task HandleAsync(TrackPushEvent @event, CancellationToken cancellationToken = default) { // 应用+客户id找推送映射配置 - var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.AppKey, @event.Customer); + var pushConfigs = await trackPushCacheService.GetPushContentConfigsAsync(@event.AppKey, @event.Customer); if (pushConfigs == null || pushConfigs.Count == 0) return; if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace()) { @@ -75,7 +56,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe // 查找所有订单 var orderIds = @event.OrderIdTrackMetadataDict.Keys.ToHashSet(); - var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => + var orders = await orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, @@ -134,8 +115,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; - var r = await _interfaceForwardService.BatchForwardThenPushToQueueAsync( - _options.CurrentValue.PushTrackInterfaceCode, + var r = await interfaceForwardService.BatchForwardThenPushToQueueAsync( + options.CurrentValue.PushTrackInterfaceCode, serviceProviderCode, items.Cast().ToList(), GlobalConst.TrackPushCompletedSubscribe, @@ -369,8 +350,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { var asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; - var r = await _interfaceForwardService.BatchForwardThenPushToQueueAsync( - _options.CurrentValue.TrackUpdateInterfaceCode, + var r = await interfaceForwardService.BatchForwardThenPushToQueueAsync( + options.CurrentValue.TrackUpdateInterfaceCode, serviceProviderCode, items.Cast().ToList(), GlobalConst.TrackPushCompletedSubscribe, @@ -448,13 +429,13 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { - var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); + var history = await trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); if (history != null) { body = TextJsonHelper.Deserialize(history.PushContent); //在推送历史中，选出当前轨迹明细一致的轨迹 - var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId); + var tracks = await trackDetailsEsRepository.GetListAsync(log.OrderId); //确定轨迹标识符 var trackIdentifies = tracks.Select(x => { @@ -619,7 +600,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { using var activity = ActivityHelper.CreateSpanActivity($"TrackPushFailedPublicTopic-{order.OrderNo}"); activity.SetTag("failedMessage", failedMessage); - _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto + broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { AppKey = @event.AppKey, IsManualPush = @event.OperationMan != null, @@ -656,7 +637,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe activity.SetTag("@event", @event.ToJson()); foreach (var pushInput in pushOrUpdateInputs) { - _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto + broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { AppKey = @event.AppKey, IsManualPush = @event.OperationMan != null, diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs similarity index 71% rename from src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs index d491de7a..461d37eb 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs @@ -1,4 +1,5 @@ -using DotNetCore.CAP; +using Fake.EventBus; +using Fake.EventBus.Distributed; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Events; @@ -13,41 +14,36 @@ using TrackCenter.Repository.Aggregates.OrderAggregate; using TrackCenter.Repository.Otel; using OrderIdentifier = TrackCenter.Application.Events.OrderIdentifier; -namespace TrackCenter.Application.CAP.EventHandlers; +namespace TrackCenter.Application.Messaging; -public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe +public class TrackUpsertEventHandler( + IOrderRepository orderRepository, + IDistributedEventBus eventBus, + ProvideQueries provideQueries, + ITrackDetailsEsRepository trackDetailsEsRepository, + IOrderOperationRecordEsRepository orderOperationRecordEsRepository, + BroadcastingService broadcastingService, + WarningSignalPublisher warningSignalPublisher) + : + IEventHandler, + IEventHandler, + ITransientDependency { - private readonly ICapPublisher _capPublisher; - private readonly IOrderRepository _orderRepository; - private readonly ProvideQueries _provideQueries; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly BroadcastingService _broadcastingService; - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - private readonly WarningSignalPublisher _warningSignalPublisher; + Task IEventHandler.HandleAsync(Event eventData, CancellationToken cancellationToken) + => eventData switch + { + TrackUpsertForRefreshEvent refresh => HandleAsync(refresh, cancellationToken), + TrackUpsertEvent upsert => HandleAsync(upsert, cancellationToken), + _ => Task.CompletedTask + }; - public TrackUpsertEventHandler( - IOrderRepository orderRepository, - ICapPublisher capPublisher, - ProvideQueries provideQueries, - ITrackDetailsEsRepository trackDetailsEsRepository, - IOrderOperationRecordEsRepository orderOperationRecordEsRepository, - BroadcastingService broadcastingService, - WarningSignalPublisher warningSignalPublisher - ) - { - _orderRepository = orderRepository; - _capPublisher = capPublisher; - _provideQueries = provideQueries; - _trackDetailsEsRepository = trackDetailsEsRepository; - _orderOperationRecordEsRepository = orderOperationRecordEsRepository; - _broadcastingService = broadcastingService; - _warningSignalPublisher = warningSignalPublisher; - } + public Task HandleAsync(TrackUpsertEvent @event, CancellationToken cancellationToken = default) + => ConsumerAsync(@event); - [UnitOfWork] - [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] - [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] - public virtual async Task ConsumerAsync(TrackUpsertEvent @event) + public Task HandleAsync(TrackUpsertForRefreshEvent @event, CancellationToken cancellationToken = default) + => ConsumerAsync(@event); + + private async Task ConsumerAsync(TrackUpsertEvent @event) { if (!CanDelete(@event)) return; @@ -56,22 +52,22 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe await HandleTrackAsync(@event, canNextOrderIds); if (canNextOrderIds.Count == 0) return; - // ????????????????? + // 根据订单关联关系异步向下刷（异步） await DownByOrderLink(@event, canNextOrderIds); } private async Task DownByOrderLink(TrackUpsertEvent @event, List canNextOrderIds) { - // ????????????????? - var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray()); + // 根据订单关联关系异步向下刷（异步） + var linkedOrders = await orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray()); foreach (var order in @event.OrderIdentifiers) { - // ??????????? + // 理论上小包在这不会分组 foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId) .GroupBy(x => new { x.AppKey, - x.Customer // ????????????????? + x.Customer // 这里按客户分组是为了后续的推送方便 })) { var nextEvent = @event.DeepCopy(); @@ -85,15 +81,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash, }; - // ??????????? + // 不同应用需要做内容映射 if (@event.AppKey != nextEvent.AppKey) { - var mapCode = (await _provideQueries.MapTrackCodesAsync(@event.AppKey, nextEvent.AppKey, @event.TrackCode)) + var mapCode = (await provideQueries.MapTrackCodesAsync(@event.AppKey, nextEvent.AppKey, @event.TrackCode)) .FirstOrDefault(); if (mapCode == null) { - // ????????????? + // 跨应用没做映射的轨迹不处理 continue; } @@ -111,7 +107,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { var subEvent = nextEvent.DeepCopy(); subEvent.OrderIdentifiers = subOrderIds.ToArray(); - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); + await eventBus.PublishAsync(subEvent); } } } @@ -119,10 +115,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private static bool CanDelete(TrackUpsertEvent @event) { - // ???????????? + // 系统创建的轨迹不允许修改 if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false }) { - Log.Warning($"??????????????{@event}", true); + Log.Warning($"系统创建的轨迹不允许被修改：{@event}", true); return false; } @@ -131,7 +127,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var dup = @event.Visited.Intersect(orderIds).ToArray(); if (dup.Any()) { - Log.Warning($"{(@event.IsAdd ? "??" : "??")}???????????{dup.JoinAsString(",")}", true); + Log.Warning($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true); return false; } @@ -140,29 +136,29 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) { - //#7001:?????????????????????????????????????? - var trackContent = await _provideQueries.GetTrackContentAsync(@event.AppKey, @event.TrackCode); + //#7001:未来潜在需求，当轨迹内容匹配预设，记录一个标记，保证下刷过程都是使用预设内容 + var trackContent = await provideQueries.GetTrackContentAsync(@event.AppKey, @event.TrackCode); if (trackContent == null) { - // ???????????? - Log.Warning($"????????{@event}", true); + // 找不到轨迹内容，直接结束 + Log.Warning($"找不到轨迹内容：{@event}", true); return; } - // ??????????????? + // 根据轨迹内容标识，确定操作权限 var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial; - // ??????? + // 不允许手动创建 if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false) { - PublicTopicAllFailed(@event, trackContent, "?????????????"); + PublicTopicAllFailed(@event, trackContent, "轨迹内容标识不允许手动创建"); return; } - // ??????? + // 不允许手动更新 if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false) { - PublicTopicAllFailed(@event, trackContent, "?????????????"); + PublicTopicAllFailed(@event, trackContent, "轨迹标识内容不允许手动更新"); return; } @@ -181,30 +177,30 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (insertTracks.Any()) { - // ??add to es tips:?????????? - await _trackDetailsEsRepository.AddManyAsync(insertTracks); + // 批量add to es tips:注意副作用填充版本号 + await trackDetailsEsRepository.AddManyAsync(insertTracks); - // es?????????? + // es新增后填充版本号字典 insertTracks.ForEach(t => orderIdTrackMetadataDict.Add(t.OrderId, TrackMetadata.Build(t.Id, t.Version, t.CreateTime))); - //?????????? - _ = _orderOperationRecordEsRepository.AddManyAsync(insertTracks.Select(t => BuildOrderOperationRecord(@event, t, new OrderIdentifier + // 批量写入订单操作日志 + _ = orderOperationRecordEsRepository.AddManyAsync(insertTracks.Select(t => BuildOrderOperationRecord(@event, t, new OrderIdentifier { OrderNo = t.OrderNo, OrderId = t.OrderId })).ToList()); - // ?????????????? - await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), + // 添加轨迹触发订单拉取状态更新 + await orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), @event.TrackCode); - // ???????????????? - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, insertTracks); + // 发布预警信号，稍后将进行预警计算 + warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, insertTracks); - // ??????????????pod?? + // 如果存在轨迹标记，则触发订单pod拉取 if (!string.IsNullOrWhiteSpace(trackContent.TrackFlagId)) { - await _capPublisher.PublishAsync(GlobalConst.OrderPodPullSubscribe, new OrderPodPullEvent + await eventBus.PublishAsync(new OrderPodPullEvent { AppKey = @event.AppKey, OrderIdEsIdDict = insertTracks.ToDictionary(x => x.OrderId, y => y.Id), @@ -213,25 +209,24 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe } } - // ???????? + // 触发轨迹变更广播 await TrackChangePublicTopicAsync(trackContent, trackChangeEtos, true); - // ?????????????? - await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, - new TrackChangeAfterUpdateOrderEvent - { - OrderIds = canNextOrderIds.ToArray(), - IsCompleted = trackContent.IsCompleted, - TrackStatus = trackContent.TrackStatus, - TrackContent = trackContent.TrackContent, - TrackEnContent = trackContent.TrackEnContent, - TrackPlace = @event.TrackPlace, - TrackTime = @event.TrackTime, - TrackTimeZone = @event.TrackTimeZone, - }); + // 触发订单主表更新事件（异步） + await eventBus.PublishAsync(new TrackChangeAfterUpdateOrderEvent + { + OrderIds = canNextOrderIds.ToArray(), + IsCompleted = trackContent.IsCompleted, + TrackStatus = trackContent.TrackStatus, + TrackContent = trackContent.TrackContent, + TrackEnContent = trackContent.TrackEnContent, + TrackPlace = @event.TrackPlace, + TrackTime = @event.TrackTime, + TrackTimeZone = @event.TrackTimeZone, + }); // 触发轨迹推送事件（异步） - // tips:???????????????? + // tips:同一个轨迹内容，多个订单多条轨迹 var trackPushEvent = new TrackPushEvent { OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, @@ -253,7 +248,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe ForcedPush = @event.ForcedPush, IsToMq = (trackContent.IsToMq || @event.IsToMq) }; - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + await eventBus.PublishAsync(trackPushEvent); } private async Task> UpsertTrackAsync( @@ -267,7 +262,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); // important：轨迹code相同，时间、发生地不同的，可能存在多条！！ - var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); + var tracks = await trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); var insertTracks = new List(); var updateTracks = new List(); @@ -290,7 +285,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe } else { - var message = $"???????EsId:[{@event.EsId}]????????????"; + var message = $"没有通过传入的EsId:[{@event.EsId}]查询到轨迹信息，更新失败"; AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); } @@ -299,21 +294,21 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList(); - // tips:??????????8?????? + // tips:如果没传时区则使用+8兜底进行过滤 var timezone = Math.Abs(@event.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 0.01f ? 8.0f : @event.TrackTimeZone; - // ?????? + // 根据时间过滤 existTracks = existTracks.Where(x => x.GetUtc0TrackTime() == @event.TrackTime.Subtract(TimeSpan.FromHours(timezone))).ToList(); - // ???????? + // 根据扩展信息过滤 existTracks = existTracks.Where(x => x.Extra == @event.Extra).ToList(); if (@event.IsAdd) { if (existTracks.Count > 0) { - const string message = "??????????????????????"; + const string message = "订单已经存在不允许重复的轨迹，不允许重复添加"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; @@ -325,10 +320,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe } else { - // ?????? + // 不存在不让改 if (existTracks.Count == 0) { - const string message = "?????????????"; + const string message = "订单不存在轨迹，不允许修改"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; @@ -336,7 +331,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (existTracks.Count > 1) { - const string message = "?????????????"; + const string message = "订单存在多条轨迹不允许修改"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; @@ -356,22 +351,22 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (updateTracks.Any()) { - // ????????? - await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks); - // ???????? updateTracks.ForEach(x => orderIdTrackMetadataDict.Add(x.OrderId, TrackMetadata.Build(x.Id, x.Version, x.CreateTime))); - // ???????? - _ = _orderOperationRecordEsRepository.AddManyAsync(operationRecords); - // ??????????????? - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); + // 批量差异化更新轨迹 + await trackDetailsEsRepository.BulkUpdateAsync(updateTracks); + // 填充轨迹元数据 + updateTracks.ForEach(x => orderIdTrackMetadataDict.Add(x.OrderId, TrackMetadata.Build(x.Id, x.Version, x.CreateTime))); + // 批量写入操作日志 + _ = orderOperationRecordEsRepository.AddManyAsync(operationRecords); + // 发布预警信号稍后将进行预警计算 + warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); } return insertTracks; } - private static TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order, TrackContentVO trackContent) { - // ?????????? ????????????? + // 非外部轨迹才允许修改 【发生地】【时间】【时区】 if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide) { track.TrackPlace = @event.TrackPlace; @@ -379,11 +374,11 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe track.TrackTimeZone = @event.TrackTimeZone; } - //?????????? + // 允许二次修改轨迹内容 track.TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content; track.TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus; - // ?????????? + // 轨迹发生地的更多信息 track.TrackPlaceCity = GlobalConst.EmptyString; track.TrackPlaceCityId = GlobalConst.NotExistId; track.TrackPlaceProvince = GlobalConst.EmptyString; @@ -392,16 +387,16 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe track.TrackPlaceNationId = GlobalConst.NotExistId; track.TrackPlacePostcode = GlobalConst.EmptyString; - // ?????????????????????? + // 签收地时区，签收地时间 track.DestinationTimeZone = null; track.EstimatedDeliveryDate = null; - // ???utc0??? + // 转换为utc0的时间 track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); - // ????ids???????????????? + // 轨迹标记ids，是否对外，备注，扩展信息，附件 track.TrackFlagIds = trackContent.TrackFlagId; track.IsExternal = @event.IsExternal ?? trackContent.IsExternal; track.Remarks = @event.Remarks ?? GlobalConst.EmptyString; @@ -412,8 +407,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe FileName = x.FileName }).ToList(); - - // ???? + // 审计属性 track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser; track.UpdateTime = DateTime.Now; @@ -424,13 +418,13 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { var track = new TrackDetailsIndex { - // ???? + // 订单信息 AppKey = @event.AppKey, OrderId = order.OrderId, OrderNo = order.OrderNo, OrderType = order.OrderType, - // ???? + // 轨迹信息 TrackCode = @event.TrackCode, TrackContentId = trackContent.TrackContentId, TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content, @@ -442,12 +436,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackTimeZone = @event.TrackTimeZone, TrackPlace = @event.TrackPlace, - // ??????? + // 服务商轨迹信息 ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString, - // ?????????? + // 轨迹发生地的更多描述 TrackPlaceCity = GlobalConst.EmptyString, TrackPlaceCityId = GlobalConst.NotExistId, TrackPlaceProvince = GlobalConst.EmptyString, @@ -455,7 +449,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackPlaceNation = GlobalConst.EmptyString, TrackPlaceNationId = GlobalConst.NotExistId, TrackPlacePostcode = GlobalConst.EmptyString, - // ????ids????????????????????? + // 轨迹标记ids，轨迹来源，是否对外，备注，扩展信息，附件 TrackFlagIds = trackContent.TrackFlagId, Source = @event.Source, IsExternal = @event.IsExternal ?? trackContent.IsExternal, @@ -467,12 +461,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe FileName = x.FileName }).ToList(), - // ???? + // 审计属性 CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now }; - // ???utc0??? + // 转换为utc0的时间 track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); @@ -497,15 +491,17 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private async Task TrackChangePublicTopicAsync(TrackContentVO trackContent, List list, bool isDelay = false) { if (list.Count == 0) return; - //??????????s????https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html - //????????????????? - //?????????? - //??1??!???????????????????????????? //??2?????????es???????? //??3???????????????? + // 理想情况在写入之后1s会刷盘 https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html + // 增加一个版本号来确定是否真的变更了 + // 没有刷盘成功事件回调 + // 方案1：!!【轮询等待静默刷盘】查出版本号确实发生了变更在发送广播 + // 方案2：考虑强制刷盘（对es压力大）后发送 + // 方案3：采用一个较大延时，不查询直接发 if (isDelay) await Task.Delay(1000); foreach (var eto in list) - _broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); + broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); } private void PublicTopicAllFailed(TrackUpsertEvent @event, TrackContentVO trackContent, string failedReason) @@ -514,7 +510,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe foreach (var orderNo in @event.OrderIdentifiers.Select(x => x.OrderNo)) { var eto = BuildEto(@event, trackContent, orderNo, null, failedReason); - _broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); + broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); } } diff --git a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs index aabd805c..5c1ed968 100644 --- a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -34,7 +34,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe); } - [UnitOfWork] + [UnitOfWork(IsTransactional = true)] public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; diff --git a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs index c74258d8..ade732fe 100644 --- a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs @@ -1,6 +1,6 @@ using System.Diagnostics; using System.Text; -using DotNetCore.CAP; +using Fake.EventBus.Distributed; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Logging; @@ -21,7 +21,7 @@ namespace TrackCenter.Application.Rabbit; [ExposeServices(typeof(IRabbitHandler))] public class TrackRefreshEventHandler( IOrderRepository orderRepository, - ICapPublisher capPublisher, + IDistributedEventBus eventBus, TrackPullCacheService trackPullCacheService, ITrackDetailsEsRepository trackDetailsEsRepository, ILogger logger, @@ -33,7 +33,7 @@ public class TrackRefreshEventHandler( return coptions.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe); } - [UnitOfWork] + [UnitOfWork(IsTransactional = true)] public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; @@ -252,7 +252,7 @@ public class TrackRefreshEventHandler( } // 新增/更新轨迹 - var trackUpsertEvent = new TrackUpsertEvent + var trackUpsertEvent = new TrackUpsertForRefreshEvent { OrderIdentifiers = new[] @@ -284,7 +284,7 @@ public class TrackRefreshEventHandler( Extra = trackExtra }; - await capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); + await eventBus.PublishAsync(trackUpsertEvent); } private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs index 255dbbfb..3371500e 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs @@ -1,4 +1,6 @@ -using DotNetCore.CAP; +using System.Data; +using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Events; using TrackCenter.Domain.Aggregates.OrderAggregate; @@ -11,8 +13,8 @@ namespace TrackCenter.Application.Services.ProvideServices; /// public partial class OrderService { - private readonly ICapPublisher _capPublisher; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; + private readonly IDistributedEventBus _eventBus = eventBus; + private readonly ITrackDetailsEsRepository _trackDetailsEsRepository = trackDetailsEsRepository; /// /// 关联、解除关联订单 @@ -21,9 +23,10 @@ public partial class OrderService /// /// [HttpPost("Order/Link")] - public async Task LinkOrderAsync(LinkOrderInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task LinkOrderAsync(LinkOrderInput input) { - var orderId = await _orderRepository.GetFirstAsync(x => + var orderId = await orderRepository.GetFirstAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, x => x.Id); if (orderId == default) @@ -60,13 +63,13 @@ public partial class OrderService { // 获取已关联的订单 var linkOrderNos = - await _orderRepository.GetLinkedOrdersAsync(orderId, input.LinkAppKey, input.LinkOrderType); + await orderRepository.GetLinkedOrdersAsync(orderId, input.LinkAppKey, input.LinkOrderType); // 计算出未关联的订单 var noLinkOrderNos = input.LinkOrderNos.Except(linkOrderNos).ToList(); // 查询存在的订单 - var existedOrders = await _orderRepository.GetListAsync(x => + var existedOrders = await orderRepository.GetListAsync(x => x.AppKey == input.LinkAppKey && x.OrderType == input.LinkOrderType && noLinkOrderNos.Contains(x.OrderNo), x => new @@ -90,7 +93,7 @@ public partial class OrderService return 0; } - await _orderRepository.LinkOrderAsync(existedOrders.Select(x => new OrderLinkEntity + await orderRepository.LinkOrderAsync(existedOrders.Select(x => new OrderLinkEntity { OrderId = orderId, LinkOrderId = x.Id, @@ -127,7 +130,7 @@ public partial class OrderService IsAdd = true // 存在则不新增 }; - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, fillTrackEvent); + await _eventBus.PublishAsync(fillTrackEvent); } } @@ -136,11 +139,11 @@ public partial class OrderService private async Task UnLinkOrderAsync(LinkOrderInput input, int orderId) { - var unLinkIds = await _orderRepository.GetLinkedOrdersAsync(orderId, input.LinkAppKey, input.LinkOrderType, + var unLinkIds = await orderRepository.GetLinkedOrdersAsync(orderId, input.LinkAppKey, input.LinkOrderType, input.LinkOrderNos); // real del - await _orderRepository.UnLinkByIdsAsync(unLinkIds); + await orderRepository.UnLinkByIdsAsync(unLinkIds); return unLinkIds.Count; } } diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.PullConfigs.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.PullConfigs.cs index 85bc64a0..4f421f0d 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.PullConfigs.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.PullConfigs.cs @@ -23,7 +23,7 @@ public partial class OrderService return new FindOrderPullConfigsOutput(); } - var pullConfigs = await _orderRepository.FindOrderPullConfigsAsync(input.AppKey, input.OrderType, orderNos); + var pullConfigs = await orderRepository.FindOrderPullConfigsAsync(input.AppKey, input.OrderType, orderNos); var foundOrderNos = pullConfigs .Select(x => x.OrderNo) .Distinct() @@ -57,7 +57,7 @@ public partial class OrderService [HttpPost("Order/UpdateOrderPullConfigs")] public async Task UpdateOrderPullConfigsAsync(UpdateOrderPullConfigsInput input) { - var orderIds = await _orderRepository.GetListAsync( + var orderIds = await orderRepository.GetListAsync( x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id); @@ -69,7 +69,7 @@ public partial class OrderService var inputVo = ObjectMapper .Map(input.Config); inputVo.OrderIds = orderIds.ToArray(); - var cnt = await _orderRepository.UpdateOrderPullConfigForServiceProviderAsync(inputVo); + var cnt = await orderRepository.UpdateOrderPullConfigForServiceProviderAsync(inputVo); Log.Warning( $"{CurrentUser.UserName}正在修改服务商拉取信息，本次共计影响了{orderIds.Count}个订单、{cnt}段拉取，请求参数是：{input.ToJson()}", diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.V2.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.V2.cs index af38574f..992449be 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.V2.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.V2.cs @@ -19,7 +19,7 @@ public partial class OrderService [VerifySignature] public async Task UpsertSmallBagAsync(UpsertSmallBagInput input) { - var order = await _orderRepository.GetFirstAsync(x => + var order = await orderRepository.GetFirstAsync(x => x.AppKey == AppKey && x.OrderType == TrackOrderType.小包 && x.OrderNo == input.OrderNo); if (order != null) { @@ -27,7 +27,7 @@ public partial class OrderService order = await BuiltOrder(input); ApplyUpsertOntoExisting(order, existing, resetTrackInfoWhenTrackNoChanged: true); - await _orderRepository.UpdateOrderAsync(order); + await orderRepository.UpdateOrderAsync(order); var index = new OrderOperationRecordIndex { @@ -39,13 +39,13 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, }; - _ = _orderOperationRecordEsRepository.AddAsync(index); + _ = orderOperationRecordEsRepository.AddAsync(index); return new IdBaseDto { Id = order.Id }; } order = await BuiltOrder(input); - var orderId = await _orderRepository.CreateOrderAsync(order); + var orderId = await orderRepository.CreateOrderAsync(order); var indexForAdd = new OrderOperationRecordIndex { @@ -57,7 +57,7 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }; - _ = _orderOperationRecordEsRepository.AddAsync(indexForAdd); + _ = orderOperationRecordEsRepository.AddAsync(indexForAdd); return new IdBaseDto { Id = orderId }; } @@ -69,7 +69,7 @@ public partial class OrderService [VerifySignature] public async Task UpsertBigBagAsync(UpsertBigBagInput input) { - var order = await _orderRepository.GetFirstAsync(x => + var order = await orderRepository.GetFirstAsync(x => x.AppKey == AppKey && x.OrderType == TrackOrderType.大包 && x.OrderNo == input.BagNo); if (order != null) { @@ -77,7 +77,7 @@ public partial class OrderService order = await BuiltOrder(input); ApplyUpsertOntoExisting(order, existing, resetTrackInfoWhenTrackNoChanged: false); - await _orderRepository.UpdateOrderAsync(order); + await orderRepository.UpdateOrderAsync(order); var index = new OrderOperationRecordIndex { @@ -89,13 +89,13 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, }; - _ = _orderOperationRecordEsRepository.AddAsync(index); + _ = orderOperationRecordEsRepository.AddAsync(index); return new IdBaseDto { Id = order.Id }; } order = await BuiltOrder(input); - var orderId = await _orderRepository.CreateOrderAsync(order); + var orderId = await orderRepository.CreateOrderAsync(order); var indexForAdd = new OrderOperationRecordIndex { @@ -107,7 +107,7 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }; - _ = _orderOperationRecordEsRepository.AddAsync(indexForAdd); + _ = orderOperationRecordEsRepository.AddAsync(indexForAdd); return new IdBaseDto { Id = orderId }; } @@ -119,7 +119,7 @@ public partial class OrderService [VerifySignature] public async Task UpsertMasterOrderAsync(UpsertMasterOrderInput input) { - var order = await _orderRepository.GetFirstAsync(x => + var order = await orderRepository.GetFirstAsync(x => x.AppKey == AppKey && x.OrderType == TrackOrderType.主单 && x.OrderNo == input.MasterNo); if (order != null) { @@ -127,7 +127,7 @@ public partial class OrderService order = await BuiltOrder(input); ApplyUpsertOntoExisting(order, existing, resetTrackInfoWhenTrackNoChanged: false); - await _orderRepository.UpdateOrderAsync(order); + await orderRepository.UpdateOrderAsync(order); var index = new OrderOperationRecordIndex { @@ -139,13 +139,13 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, }; - _ = _orderOperationRecordEsRepository.AddAsync(index); + _ = orderOperationRecordEsRepository.AddAsync(index); return new IdBaseDto { Id = order.Id }; } order = await BuiltOrder(input); - var orderId = await _orderRepository.CreateOrderAsync(order); + var orderId = await orderRepository.CreateOrderAsync(order); var indexForAdd = new OrderOperationRecordIndex { @@ -157,7 +157,7 @@ public partial class OrderService CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }; - _ = _orderOperationRecordEsRepository.AddAsync(indexForAdd); + _ = orderOperationRecordEsRepository.AddAsync(indexForAdd); return new IdBaseDto { Id = orderId }; } @@ -183,7 +183,7 @@ public partial class OrderService ArrivalEmail = input.Consignee.Email, }; - var config = await _trackPullConfigRepository + var config = await trackPullConfigRepository .FirstOrDefaultAsync(x => x.AppKey == AppKey && x.ServiceProviderCode == input.ApiCode && x.OrderType == TrackOrderType.小包); @@ -220,7 +220,7 @@ public partial class OrderService return order; } - var config = await _trackPullConfigRepository + var config = await trackPullConfigRepository .FirstOrDefaultAsync(x => x.AppKey == AppKey && x.ServiceProviderCode == input.ApiCode && x.OrderType == TrackOrderType.大包); @@ -250,7 +250,7 @@ public partial class OrderService order.OrderFlights = new List(); - var config = await _trackPullConfigRepository + var config = await trackPullConfigRepository .FirstOrDefaultAsync(x => x.AppKey == AppKey && x.ServiceProviderCode == input.ApiCode && x.OrderType == TrackOrderType.小包); diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs index 4c733266..9211d0e2 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs @@ -1,4 +1,4 @@ -using DotNetCore.CAP; +using Fake.EventBus.Distributed; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; @@ -13,30 +13,15 @@ namespace TrackCenter.Application.Services.ProvideServices; /// [ApiExplorerSettings(GroupName = "OpenApi")] [Route("/ProviderService")] -public partial class OrderService : ApplicationServiceBase +public partial class OrderService( + IOrderRepository orderRepository, + IOrderOperationRecordEsRepository orderOperationRecordEsRepository, + ITrackPullConfigRepository trackPullConfigRepository, + IDistributedEventBus eventBus, + ITrackDetailsEsRepository trackDetailsEsRepository, + IServiceProvider serviceProvider) + : ApplicationServiceBase { - private readonly IOrderRepository _orderRepository; - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - private readonly ITrackPullConfigRepository _trackPullConfigRepository; - private readonly IServiceProvider _serviceProvider; - - public OrderService( - IOrderRepository orderRepository, - IOrderOperationRecordEsRepository orderOperationRecordEsRepository, - ITrackPullConfigRepository trackPullConfigRepository, - ICapPublisher capPublisher, - ITrackDetailsEsRepository trackDetailsEsRepository, - IServiceProvider serviceProvider - ) - { - _orderRepository = orderRepository; - _orderOperationRecordEsRepository = orderOperationRecordEsRepository; - _trackPullConfigRepository = trackPullConfigRepository; - _serviceProvider = serviceProvider; - _trackDetailsEsRepository = trackDetailsEsRepository; - _capPublisher = capPublisher; - } - /// /// 删除订单 /// @@ -46,14 +31,14 @@ public partial class OrderService : ApplicationServiceBase [HttpPost("Order/Delete")] public async Task DeleteOrderAsync(DeleteOrderInput input) { - var order = await _orderRepository.GetFirstAsync(x => + var order = await orderRepository.GetFirstAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo); if (order == null) { throw ProviderServiceActionResult.Exception("删除订单失败", $"订单：{input.OrderNo}不存在"); } - await _orderRepository.DeleteOrderAsync(order.Id); + await orderRepository.DeleteOrderAsync(order.Id); // 预警计算 WarningCalc_OrderDeletedOrCompleted(order); @@ -66,7 +51,7 @@ public partial class OrderService : ApplicationServiceBase order.IsDeleted = true; _ = Task.Run(async () => { - var scope = _serviceProvider.CreateAsyncScope(); + var scope = serviceProvider.CreateAsyncScope(); try { var handler = scope.ServiceProvider.GetRequiredService(); diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackPullService.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackPullService.cs index a4248da4..031e0c06 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackPullService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackPullService.cs @@ -15,21 +15,12 @@ namespace TrackCenter.Application.Services.ProvideServices; /// [ApiExplorerSettings(GroupName = "OpenApi")] [Route("/ProviderService")] -public class TrackPullService : ApplicationServiceBase +public class TrackPullService( + ProvideQueries provideQueries, + InterfaceForwardService interfaceForwardService, + IOptionsMonitor options) + : ApplicationServiceBase { - private readonly IOptionsMonitor _options; - private readonly ProvideQueries _provideQueries; - private readonly InterfaceForwardService _interfaceForwardService; - - public TrackPullService(ProvideQueries provideQueries, - InterfaceForwardService interfaceForwardService, - IOptionsMonitor options) - { - _provideQueries = provideQueries; - _interfaceForwardService = interfaceForwardService; - _options = options; - } - /// /// 手动刷新轨迹 /// @@ -46,7 +37,7 @@ public class TrackPullService : ApplicationServiceBase // tips：这里没有区分订单类型，就算的不同类型的单一起刷新了也无所谓 var (orders, skipReasons) = - await _provideQueries.DiagnoseManualRefreshAsync(input.AppKey, input.OrderNos); + await provideQueries.DiagnoseManualRefreshAsync(input.AppKey, input.OrderNos); if (orders.Count == 0) { @@ -79,8 +70,8 @@ public class TrackPullService : ApplicationServiceBase using (Track2InterfaceRelayAppContext.Use( new Track2InterfaceRelayAppContext(group.Key.AppKey))) { - var res = await _interfaceForwardService.BatchForwardThenPushToQueueAsync( - _options.CurrentValue.GetTrackInterfaceCode, serviceProviderCode, data, + var res = await interfaceForwardService.BatchForwardThenPushToQueueAsync( + options.CurrentValue.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs index 4afc5335..f371e4ba 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs @@ -1,4 +1,6 @@ -using DotNetCore.CAP; +using System.Data; +using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Dtos.TrackOrder; using TrackCenter.Application.Dtos.TrackPush; @@ -18,20 +20,20 @@ namespace TrackCenter.Application.Services.ProvideServices; [Route("/ProviderService")] public class TrackPushService : ApplicationServiceBase { - private readonly ICapPublisher _capPublisher; + private readonly IDistributedEventBus _eventBus; private readonly IOrderRepository _orderRepository; private readonly ITrackPushConfigRepository _trackPushConfigRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; public TrackPushService(IOrderRepository orderRepository - , ICapPublisher capPublisher + , IDistributedEventBus eventBus , ITrackPushRecordEsRepository trackPushRecordEsRepository , ITrackPushConfigRepository trackPushConfigRepository, ITrackDetailsEsRepository trackDetailsEsRepository) { _orderRepository = orderRepository; - _capPublisher = capPublisher; + _eventBus = eventBus; _trackPushRecordEsRepository = trackPushRecordEsRepository; _trackPushConfigRepository = trackPushConfigRepository; _trackDetailsEsRepository = trackDetailsEsRepository; @@ -43,7 +45,8 @@ public class TrackPushService : ApplicationServiceBase /// /// [HttpPost("Track/Push")] - public async Task PushTrackAsync(PushTrackInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task PushTrackAsync(PushTrackInput input) { input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); if (input.OrderNos.Count == 0) @@ -152,7 +155,7 @@ public class TrackPushService : ApplicationServiceBase } }; - await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent); + await _eventBus.PublishAsync(trackPushEvent); pushedOrderNos.Add(order.OrderNo); } diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackService.Query.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackService.Query.cs index 44f58896..26b44fca 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackService.Query.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackService.Query.cs @@ -13,7 +13,7 @@ namespace TrackCenter.Application.Services.ProvideServices; /// public partial class TrackService { - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; + private readonly ITrackDetailsEsRepository _trackDetailsEsRepository = trackDetailsEsRepository; private readonly IHttpContextAccessor _contextAccessor; private IObjectStorage ObjectStorage => LazyServiceProvider.GetRequiredService(); diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs index 70189343..0a4d22e0 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs @@ -1,4 +1,6 @@ -using DotNetCore.CAP; +using System.Data; +using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Events; @@ -15,34 +17,23 @@ namespace TrackCenter.Application.Services.ProvideServices; /// [ApiExplorerSettings(GroupName = "OpenApi")] [Route("/ProviderService")] -public partial class TrackService : ApplicationServiceBase +public partial class TrackService( + IOrderRepository orderRepository, + IDistributedEventBus eventBus, + ITrackDetailsEsRepository trackDetailsEsRepository, + ITrackContentRepository trackContentRepository) + : ApplicationServiceBase { - private readonly ICapPublisher _capPublisher; - private readonly IOrderRepository _orderRepository; - private readonly ITrackContentRepository _trackContentRepository; - - public TrackService( - IOrderRepository orderRepository, - ICapPublisher capPublisher, - ITrackDetailsEsRepository trackDetailsEsRepository, - ITrackContentRepository trackContentRepository - ) - { - _orderRepository = orderRepository; - _capPublisher = capPublisher; - _trackContentRepository = trackContentRepository; - _trackDetailsEsRepository = trackDetailsEsRepository; - } - /// /// 删除轨迹 /// /// /// [HttpPost("Track/Delete")] - public async Task DeleteTrackAsync(DeleteTrackInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task DeleteTrackAsync(DeleteTrackInput input) { - var orderId = await _orderRepository.GetFirstAsync(x => + var orderId = await orderRepository.GetFirstAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, @@ -99,7 +90,7 @@ public partial class TrackService : ApplicationServiceBase } // 删除轨迹(异步) - await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, new TrackDeleteEvent + await eventBus.PublishAsync(new TrackDeleteEvent { EsIds = esIds, OrderIds = new[] { orderId }, @@ -118,10 +109,11 @@ public partial class TrackService : ApplicationServiceBase /// /// [HttpPost("Track/BatchDelete")] - public async Task BatchDeleteTrackAsync(BatchDeleteTrackInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task BatchDeleteTrackAsync(BatchDeleteTrackInput input) { input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空 - var orders = await _orderRepository.GetListAsync(x => + var orders = await orderRepository.GetListAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), @@ -173,7 +165,7 @@ public partial class TrackService : ApplicationServiceBase AppKey = input.AppKey, OperationMan = CurrentUser.UserName }; - await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event); + await eventBus.PublishAsync(@event); } /// @@ -182,11 +174,12 @@ public partial class TrackService : ApplicationServiceBase /// /// [HttpPost("Track/Upsert")] - public async Task UpsertTrackAsync(UpsertTrackInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task UpsertTrackAsync(UpsertTrackInput input) { using var activity = ActivityHelper.CreateSpanActivity("UpsertTrack"); activity.SetTag("UpsertTrackInput", input.ToJson()); - var order = await _orderRepository.GetFirstAsync(x => + var order = await orderRepository.GetFirstAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, @@ -226,7 +219,7 @@ public partial class TrackService : ApplicationServiceBase } }; @event.OperationMan = CurrentUser.UserName; - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event); + await eventBus.PublishAsync(@event); } /// @@ -235,12 +228,13 @@ public partial class TrackService : ApplicationServiceBase /// /// [HttpPost("Track/BatchUpsert")] - public async Task BatchUpsertTrackAsync(BatchUpsertTrackInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task BatchUpsertTrackAsync(BatchUpsertTrackInput input) { using var activity = ActivityHelper.CreateSpanActivity("BatchUpsert"); activity.SetTag("BatchUpsertTrackInput", input.ToJson()); input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); - var orders = await _orderRepository.GetListAsync(x => + var orders = await orderRepository.GetListAsync(x => x.AppKey == input.AppKey && x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), @@ -298,7 +292,7 @@ public partial class TrackService : ApplicationServiceBase ForcedPush = input.ForcedPush, IsToMq = input.IsToMq, }; - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event); + await eventBus.PublishAsync(@event); } } @@ -339,7 +333,7 @@ public partial class TrackService : ApplicationServiceBase #region 轨迹内容中的配置规则检查 - var trackContent = await _trackContentRepository.GetByCodeAsync(appKey, trackCode); + var trackContent = await trackContentRepository.GetByCodeAsync(appKey, trackCode); if (trackContent == null) { throw ProviderServiceActionResult.Exception("校验失败", $"轨迹代码：{trackCode}轨迹内容不存在"); diff --git a/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs b/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs index e8c7818b..8aeb3006 100644 --- a/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs +++ b/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs @@ -28,14 +28,14 @@ public class TrackPullTask( ProvideQueries provideQueries, InterfaceForwardService interfaceForwardService, IOptions interfaceOptions) - : ApplicationServiceBase, ITransientDependency + : ApplicationServiceBase { /// /// 定时任务刷新轨迹 /// /// [DisableConcurrentExecution(timeoutInSeconds: 1800)] - [UnitOfWork] + [UnitOfWork(IsTransactional = true)] [HttpPost("/TimerTask/Refresh")] public virtual async Task RefreshForTaskAsync() { @@ -66,7 +66,7 @@ public class TrackPullTask( return true; } - [UnitOfWork] + [UnitOfWork(IsTransactional = true)] public virtual async Task RefreshForConfigAsync(TrackPullConfigVo config) { var redisKey = trackPullCacheService.TrackPullKey(config); @@ -89,7 +89,7 @@ public class TrackPullTask( } } - protected async Task DoRefreshAsync(TrackPullConfigVo config, ProvideQueries provideQueries, + protected async Task DoRefreshAsync(TrackPullConfigVo config, ProvideQueries queries, InterfaceForwardService interfaceForward, InterfaceRelayParamsConfigOptions interfaceOptionsValue) { @@ -97,7 +97,7 @@ public class TrackPullTask( { var page = 1; var orders = - await provideQueries.FindOrderWithServiceProviderAsync(config, page); + await queries.FindOrderWithServiceProviderAsync(config, page); while (orders.Any()) { var serviceProviderCode = orders.First().ServiceProviderCode; @@ -122,7 +122,7 @@ public class TrackPullTask( } } - orders = await provideQueries.FindOrderWithServiceProviderAsync(config, ++page); + orders = await queries.FindOrderWithServiceProviderAsync(config, ++page); } } } diff --git a/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs b/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs index f9c05bb0..9faaebcb 100644 --- a/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs +++ b/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs @@ -21,29 +21,16 @@ namespace TrackCenter.Application.Services.TrackOrders; /// [ApiExplorerSettings(GroupName = "轨迹订单")] [Route("/Track/Order")] -public class TrackOrderService : ApplicationServiceBase +public class TrackOrderService( + IOrderRepository orderRepository, + ITrackDetailsEsRepository trackDetailsEsRepository, + ITrackPushRecordEsRepository trackPushRecordEsRepository, + IOrderOperationRecordEsRepository orderOperationRecordEsRepository, + IServiceProvider serviceProvider) + : ApplicationServiceBase { - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - private readonly IOrderRepository _orderRepository; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; - private readonly IServiceProvider _serviceProvider; private IObjectStorage ObjectStorage => LazyServiceProvider.GetRequiredService(); - public TrackOrderService( - IOrderRepository orderRepository, - ITrackDetailsEsRepository trackDetailsEsRepository, - ITrackPushRecordEsRepository trackPushRecordEsRepository, - IOrderOperationRecordEsRepository orderOperationRecordEsRepository, - IServiceProvider serviceProvider) - { - _orderRepository = orderRepository; - _trackDetailsEsRepository = trackDetailsEsRepository; - _trackPushRecordEsRepository = trackPushRecordEsRepository; - _orderOperationRecordEsRepository = orderOperationRecordEsRepository; - _serviceProvider = serviceProvider; - } - #region 轨迹订单列表 /// @@ -55,7 +42,7 @@ public class TrackOrderService : ApplicationServiceBase public async Task> GetTrackOrderPagedList(GetPageTrackOrderInput input) { var vo = ObjectMapper.Map(input); - var pageList = await _orderRepository.GetTrackOrderPagedList(vo); + var pageList = await orderRepository.GetTrackOrderPagedList(vo); return pageList; } @@ -70,7 +57,7 @@ public class TrackOrderService : ApplicationServiceBase [HttpPost("UpdateOrderStatus")] public async Task UpdateOrderStatus(UpdateOrderStatusInput input) { - var order = await _orderRepository.GetFirstAsync(x => x.Id == input.OrderId); + var order = await orderRepository.GetFirstAsync(x => x.Id == input.OrderId); if (order == null) throw new BusinessException(message: "订单不存在"); @@ -81,7 +68,7 @@ public class TrackOrderService : ApplicationServiceBase } order.IsCompleted = input.IsCompleted; - var b = await _orderRepository.UpdateColumnAsync(order, x => new + var b = await orderRepository.UpdateColumnAsync(order, x => new { x.IsCompleted }); @@ -101,7 +88,7 @@ public class TrackOrderService : ApplicationServiceBase [HttpGet("GetTrackPushRecords")] public async Task> GetTrackPushRecordsAsync(int orderId) { - var records = await _trackPushRecordEsRepository.GetTrackPushRecordsAsync(orderId); + var records = await trackPushRecordEsRepository.GetTrackPushRecordsAsync(orderId); var res = ObjectMapper.Map, List>(records); @@ -139,13 +126,13 @@ public class TrackOrderService : ApplicationServiceBase if (whereExpression == default) throw new BusinessException(message: "请求参数异常"); - var order = await _orderRepository.GetFirstAsync(whereExpression); + var order = await orderRepository.GetFirstAsync(whereExpression); if (order == default) { throw new BusinessException(message: "订单不存在"); } - var trackDetails = await _trackDetailsEsRepository.GetListAsync(order.Id); + var trackDetails = await trackDetailsEsRepository.GetListAsync(order.Id); if (trackDetails.IsNullOrEmpty()) { return new List(); @@ -209,11 +196,11 @@ public class TrackOrderService : ApplicationServiceBase { var res = new List(); - var order = await _orderRepository.GetFirstAsync(m => m.OrderNo == input.OrderNumber); + var order = await orderRepository.GetFirstAsync(m => m.OrderNo == input.OrderNumber); if (order == null) throw new BusinessException("订单不存在"); - var trackDetails = await _trackDetailsEsRepository.GetListAsync(order.Id); + var trackDetails = await trackDetailsEsRepository.GetListAsync(order.Id); //时间轴不显示轨迹内容为空的轨迹，进行过滤 trackDetails = trackDetails.Where(x => !x.TrackContent.IsNullOrWhiteSpace() && x.TrackContent != GlobalConst.EmptyString).ToList(); @@ -278,7 +265,7 @@ public class TrackOrderService : ApplicationServiceBase [HttpPost("DeleteTrackById")] public async Task DeleteTrackByIdAsync(string trackId) { - var track = await _trackDetailsEsRepository.GetByIdAsync(trackId); + var track = await trackDetailsEsRepository.GetByIdAsync(trackId); // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract if (track == null) @@ -286,7 +273,7 @@ public class TrackOrderService : ApplicationServiceBase throw new BusinessException(message: "轨迹不存在！"); } - await _trackDetailsEsRepository.DeleteTrackByIdAsync(trackId, track.IndexName); + await trackDetailsEsRepository.DeleteTrackByIdAsync(trackId, track.IndexName); return true; } @@ -298,14 +285,14 @@ public class TrackOrderService : ApplicationServiceBase [HttpGet("GetOrderOperationsById")] public async Task> GetOrderOperationsByIdAsync(int orderId) { - return await _orderOperationRecordEsRepository.GetOrderOperationsByIdAsync(orderId); + return await orderOperationRecordEsRepository.GetOrderOperationsByIdAsync(orderId); } private void WarningCalc_OrderDeletedOrCompleted(OrderEntity order) { _ = Task.Run(async () => { - var scope = _serviceProvider.CreateAsyncScope(); + var scope = serviceProvider.CreateAsyncScope(); try { var handler = scope.ServiceProvider.GetRequiredService(); diff --git a/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs b/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs index c144a493..368d0c6d 100644 --- a/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs +++ b/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs @@ -1,5 +1,7 @@ -using DotNetCore.CAP; +using System.Data; +using Fake.EventBus.Distributed; using Fake.Domain.Dtos; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.Warning; using TrackCenter.Application.Dtos.Warning.Return; @@ -25,7 +27,7 @@ public class ReturnWarningService : ApplicationServiceBase private readonly IReturnWarningHandler _returnWarningHandler; private readonly IWarningRuleRepository _warningRuleRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; - private readonly ICapPublisher _capPublisher; + private readonly IDistributedEventBus _eventBus; private readonly TrackStatusNameDictCache _trackStatusNameDictCache; private readonly AppNameDictCache _appNameDictCache; @@ -34,7 +36,7 @@ public class ReturnWarningService : ApplicationServiceBase IReturnWarningHandler returnWarningHandler, IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, - ICapPublisher capPublisher, + IDistributedEventBus eventBus, TrackStatusNameDictCache trackStatusNameDictCache, AppNameDictCache appNameDictCache ) @@ -43,7 +45,7 @@ public class ReturnWarningService : ApplicationServiceBase _returnWarningHandler = returnWarningHandler; _warningRuleRepository = warningRuleRepository; _warningDetailsEsRepository = warningDetailsEsRepository; - _capPublisher = capPublisher; + _eventBus = eventBus; _trackStatusNameDictCache = trackStatusNameDictCache; _appNameDictCache = appNameDictCache; } @@ -220,7 +222,8 @@ public class ReturnWarningService : ApplicationServiceBase /// /// [HttpPost("ReturnPush")] - public async Task ReturnPush(ReturnPushInput input) + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] + public virtual async Task ReturnPush(ReturnPushInput input) { if (input.WarningDetailIds.Count == 0) { @@ -276,7 +279,7 @@ public class ReturnWarningService : ApplicationServiceBase } }; - await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent); + await _eventBus.PublishAsync(trackPushEvent); } return $"即将进行{filtered.Count}个回传推送！"; diff --git a/src/TrackCenter.Application/TrackCenter.Application.csproj b/src/TrackCenter.Application/TrackCenter.Application.csproj index d1d3f4ba..ccab7abc 100644 --- a/src/TrackCenter.Application/TrackCenter.Application.csproj +++ b/src/TrackCenter.Application/TrackCenter.Application.csproj @@ -1,4 +1,4 @@ - + @@ -10,6 +10,7 @@ + @@ -19,12 +20,6 @@ - - - - - - diff --git a/src/TrackCenter.Application/TrackCenterApplicationModule.cs b/src/TrackCenter.Application/TrackCenterApplicationModule.cs index 0137323b..4c0de9ef 100644 --- a/src/TrackCenter.Application/TrackCenterApplicationModule.cs +++ b/src/TrackCenter.Application/TrackCenterApplicationModule.cs @@ -1,8 +1,5 @@ using System.Diagnostics; using System.Net; -using DotNetCore.CAP; -using DotNetCore.CAP.Internal; -using DotNetCore.CAP.Messages; using Fake; using Fake.EventBus.RabbitMQ; using Fake.Modularity; @@ -16,9 +13,7 @@ using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using RabbitMQ.Client; -using Savorboard.CAP.InMemoryMessageQueue; using TrackCenter.Application.Kafka; -using TrackCenter.Application.CAP; using TrackCenter.Application.HostServices; using TrackCenter.Application.Options; using TrackCenter.Application.Rabbit; @@ -41,62 +36,6 @@ public class TrackCenterApplicationModule : FakeModule options.ScanProfiles(); }); - var rabbitOptions = configuration.GetSection("RabbitMQ").Get()!.Default; - - #region CAP - - context.Services.AddSingleton(); - if (!Debugger.IsAttached) - { - context.Services.AddCap(options => - { - options.UseMySql(configuration["ConnectionStrings:Default"]!); - options.UseRabbitMQ(configure => - { - configure.HostName = rabbitOptions.HostName; - configure.UserName = rabbitOptions.UserName; - configure.Password = rabbitOptions.Password; - configure.Port = rabbitOptions.Port; - configure.BasicQosOptions = - new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount")); - }); - options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); - options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); - options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); - options.FallbackWindowLookbackSeconds = - configuration.GetValue("CAP:FallbackWindowLookbackSeconds"); - - options.UseDashboard(dashboardOptions => { dashboardOptions.PathBase = configuration["ASPNETCORE_PATHBASE"]; }); - options.FailedThresholdCallback = faileCalback => - { - try - { - Log.Error($@" - |-CAP失败回调事件： - |-处理类型：{(faileCalback.MessageType == MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")} - |-头部信息：{faileCalback.Message.Headers.ToJson()}"); - } - catch (Exception ex) - { - Log.Error("【严重】CAP失败回调事件发生未处理异常" + ex); - } - }; - }).AddSubscribeFilter(); - } - else - { - //是本地调试环境，用内存存储和内存队列 - context.Services.AddCap(capOptions => - { - capOptions.UseInMemoryStorage(); - capOptions.UseInMemoryMessageQueue(); - capOptions.UseDashboard(); - }).AddSubscribeFilter(); - context.Services.GetInitLogger().LogInformation("-- 当前为本地环境，用内存存储和内存队列"); - } - - #endregion - var redisOptions = configuration.GetSection("Redis").Get(); if (redisOptions == null) throw new FakeException("未找到Redis相关配置，请检查配置文件中是否包含名为Redis的节点");