using System.Text; using Fake.UnitOfWork; using Newtonsoft.Json; using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Helpers; using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackPushAggregate; namespace TrackCenter.Application.Rabbit; [ExposeServices(typeof(IRabbitHandler))] public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler { private readonly IOrderRepository _orderRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly BroadcastingService _broadcastingService; private readonly IReturnWarningHandler _returnWarningHandler; public TrackPushCompletedEventHandler(IOrderRepository orderRepository, ITrackPushRecordEsRepository trackPushRecordEsRepository, BroadcastingService broadcastingService, IReturnWarningHandler returnWarningHandler) { _orderRepository = orderRepository; _trackPushRecordEsRepository = trackPushRecordEsRepository; _broadcastingService = broadcastingService; _returnWarningHandler = returnWarningHandler; } public bool Enable(ConsumeOptions options) { return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe); } [UnitOfWork] public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; try { msg = Encoding.UTF8.GetString(args.Body.ToArray()); var @event = System.Text.Json.JsonSerializer.Deserialize(msg); if (@event == null) { Log.Warning( $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } await ConsumerAsync(@event); } catch (Exception ex) { Log.Error($"TrackPushCompletedEventHandler出错,错误信息:{ex}\r\n-----\r\n消息:{msg}"); } } public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { var optionType = Base64Helper.DeSerialize(@event.Attach); var pushInputs = new List(); var updateInputs = new List(); var isAdd = false; // 有多少个订单推送,就有多少个TrackPushInput/TrackUpdateInput返回 // tips:以下处理认为只处理一个单的这次推送的情况 if (optionType == OptionType.新增.ToString()) { pushInputs = JsonConvert.DeserializeObject>(@event.RequestParams)!; isAdd = true; } else if (optionType == OptionType.修改.ToString()) { updateInputs = JsonConvert.DeserializeObject>(@event.RequestParams)!; } var logJsons = pushInputs.Count != 0 ? pushInputs.Select(x => x.Log) : updateInputs.Select(x => x.Log); var logs = new List(pushInputs.Count + updateInputs.Count); foreach (var logJson in logJsons) { if (logJson.IsNullOrEmpty()) throw new InvalidOperationException(); var log = JsonConvert.DeserializeObject(logJson!); if (log == null) throw new BusinessException(message: "轨迹推送返回结果解析异常"); log.IsSuccess = @event.IsSuccess; log.Message = @event.Message; log.RequestId = @event.RequestId; logs.Add(log); } // 推送广播 if (logs.First().IsToMq) _broadcastingService.TrackPushCompletedPublicTopic(logs.Select(log => BuildDto(log, @event.Data)).ToList()); // 写入推送日志 await _trackPushRecordEsRepository.AddManyAsync(logs); // 同步更新订单主表 await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs); // 引发预警计算 await _returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs); } private async Task TrackPushAfterUpdateOrderAsync(bool isAdd, bool isSuccess, List logs) { // 修改跳过 if (!isAdd) return; if (!isSuccess) { // 最近一条推送失败 var orders = logs.Select(x => new OrderEntity { Id = x.OrderId, QuestionType = TrackProblemType.Error, LatestPushError = x.Message }).ToList(); await _orderRepository.UpdateRangeAsync(orders, x => new { x.QuestionType, x.LatestPushError }); } else { // 还原成正常态 var orders = logs.Select(x => new OrderEntity { Id = x.OrderId, QuestionType = (int) TrackProblemType.None, LatestPushError = null, LatestPushTrackCode = x.TrackCode, LatestPushTime = x.CreateTime, }).ToList(); await _orderRepository.UpdateRangeAsync(orders, x => new { x.QuestionType, x.LatestPushError, x.LatestPushTrackCode, x.LatestPushTime, }); } } private static TrackPushRecordDto BuildDto(TrackPushRecordIndex log, string interfaceRelayData) { return new TrackPushRecordDto { AppKey = log.AppKey, RequestId = log.RequestId, IsSuccess = log.IsSuccess, Message = log.Message, OperationMan = log.CreateUser, PushTime = DateTime.Now, OptionType = log.OptionType, IsManualPush = !string.Equals(log.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户,则认为是人推送 PushContent = log.PushContent, OrderNo = log.OrderNo, TrackStatus = log.TrackStatus, TrackContent = log.TrackContent, TrackTimeZone = log.TrackTimeZone, TrackTime = log.TrackTime, TrackPlace = log.TrackPlace, TrackCode = log.TrackCode, Extra = log.Extra, InterfaceRelayData = interfaceRelayData }; } }