using System.Text; using BSI.Stack.Rabbit; using BSI.Stack.Rabbit.Handler; using Newtonsoft.Json; using RabbitMQ.Client.Events; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Helpers; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackContentAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Aggregates.TrackStatusAggregate; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.Rabbit; public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler { private readonly IOrderRepository _orderRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly ITrackStatusRepository _trackStatusRepository; private readonly ITrackContentRepository _trackContentRepository; private readonly RabbitClientBus _rabbitClientBus; public TrackPushCompletedEventHandler(IOrderRepository orderRepository, ITrackPushRecordEsRepository trackPushRecordEsRepository, ITrackDetailsEsRepository trackDetailsEsRepository, ITrackStatusRepository trackStatusRepository, ITrackContentRepository trackContentRepository, RabbitClientBus rabbitClientBus) { _orderRepository = orderRepository; _trackPushRecordEsRepository = trackPushRecordEsRepository; _trackDetailsEsRepository = trackDetailsEsRepository; _trackStatusRepository = trackStatusRepository; _trackContentRepository = trackContentRepository; _rabbitClientBus = rabbitClientBus; } public ConsumeOptions Options() { return new ConsumeOptions() { Queue = GlobalConst.TrackPushCompletedSubscribe }; } public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) { var msg = Encoding.UTF8.GetString(args.Body.ToArray()); var @event = System.Text.Json.JsonSerializer.Deserialize(msg); if (@event == null) { LogHelper.Warn( $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } string? logStr = null; var optionType = Base64Helper.DeSerialize(@event.Attach); if (optionType == OptionType.新增.ToString()) { logStr = JsonConvert.DeserializeObject>(@event.RequestParams)!.First().Log; } else if (optionType == OptionType.修改.ToString()) { logStr = JsonConvert.DeserializeObject>(@event.RequestParams)!.First().Log;; } if (logStr.IsNullOrEmpty()) throw new InvalidOperationException(); var log = JsonConvert.DeserializeObject(logStr!); if (log == null) throw new BusinessException(message: "轨迹推送返回结果解析异常"); log.IsSuccess = @event.IsSuccess; log.Message = @event.Message; log.RequestId = @event.RequestId; // todo:这里不需要查es,log里面有 //查出推送轨迹,写入推送广播 var tracks = await _trackDetailsEsRepository .GetListAsync(new[] { log.OrderId }, new List { log.TrackCode }); // 推送广播 await PublicTrackPushTopicAsync(tracks.First(), log, !log.IsSuccess, log.Message); // 往es 写推送日志 await _trackPushRecordEsRepository.AddAsync(log); // 同步去更新订单主表 await TrackPushAfterUpdateOrderAsync(log); } private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log) { // tips:只有新增会影响推送进度? if (log.OptionType != OptionType.新增.ToString()) return; var order = new OrderEntity { Id = log.OrderId }; /* * 异常:最近一条推送是失败 */ if (!log.IsSuccess) { order.QuestionType = (int) TrackProblemType.Error; order.LatestPushError = log.Message; await _orderRepository.UpdateColumnAsync(order, x => new { x.QuestionType, x.LatestPushError }); } else { // 还原成正常态 order.QuestionType = (int) TrackProblemType.None; order.LatestPushError = null; order.LatestPushTrackCode = log.TrackCode; order.LatestPushTime = log.CreateTime; await _orderRepository.UpdateColumnAsync(order, x => new { x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError }); } } /// /// 轨迹推送广播 /// private async Task PublicTrackPushTopicAsync(TrackDetailsIndex track, TrackPushRecordIndex pushRecord, bool isFailure = false, string? errorMsg = null) { // view by xiaolipro at 2024-12-24 // todo:热路径下进行数据库查询影响性能,建议减少或缓存,或异步处理@王云龙 var statusList = await _trackStatusRepository.GetListAsync(m => m.SubSystemId == track.SubSystemId); if (statusList == null) return; var contentList = await _trackContentRepository.GetTrackContentBySyetemIdAsync(track.SubSystemId); if (contentList == null) return; var status = statusList.FirstOrDefault(m => m.Code == track.TrackStatusCode); if (status == null || !status.IsToMq) return; var content = contentList.FirstOrDefault(m => m.TrackContentCode == track.TrackCode); var data = new TrackPushEto { Option = pushRecord.OptionType == OptionType.新增.ToString() ? OptionType.新增 : OptionType.修改, IsManualPush = !string.Equals(pushRecord.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户,则认为是仍推送 IsFailure = isFailure, FailureReason = errorMsg, PushTime = DateTime.Now, OperationMan = pushRecord.CreateUser, EstimatedDeliveryDate = track.EstimatedDeliveryDate, TrackPlacePostcode = track.TrackPlacePostcode, TrackPlaceNation = track.TrackPlaceNation, TrackPlaceNationId = track.TrackPlaceNationId, TrackPlaceProvince = track.TrackPlaceProvince, TrackPlaceProvinceId = track.TrackPlaceProvinceId, TrackPlaceCity = track.TrackPlaceCity, TrackPlaceCityId = track.TrackPlaceCityId, Files = track.Files?.Select(x => new FileItemDto { OssKey = x.OssKey, FileName = x.FileName }).ToList(), Remarks = track.Remarks, Source = track.Source, IsVisibility = track.IsVisibility, TrackTimeZone = track.TrackTimeZone, TrackTime = track.TrackTime, TrackPlace = track.TrackPlace, TrackStatusCode = track.TrackStatusCode, EnStatus = track.TrackEnContent, Content = track.TrackContent, TrackCode = track.TrackCode, SubSystemId = track.SubSystemId, OrderNo = track.OrderNo, OrderIdentifiers = new OrderIdentifier[] { new OrderIdentifier { OrderId = track.OrderId, OrderNo = track.OrderNo, OrderType = track.OrderType } }, Extra = track.Extra }; _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); } }