From b6a2cc8615be3d5afbd4b7fa36b702344c53ed64 Mon Sep 17 00:00:00 2001 From: hardstylewyl <827778732@qq.com> Date: Tue, 24 Dec 2024 15:09:12 +0800 Subject: [PATCH] =?UTF-8?q?update:=20=E4=BC=98=E5=8C=96=E8=BD=A8=E8=BF=B9?= =?UTF-8?q?=E7=9B=B8=E5=85=B3=E5=B9=BF=E6=92=AD,=E6=8A=BD=E7=A6=BB?= =?UTF-8?q?=E5=B9=BF=E6=92=AD=E4=B8=BABroadcastingService=EF=BC=8C?= =?UTF-8?q?=E6=B6=89=E5=8F=8A=E8=BD=A8=E8=BF=B9=E5=8F=98=E6=9B=B4=EF=BC=8C?= =?UTF-8?q?=E8=BD=A8=E8=BF=B9=E6=8E=A8=E9=80=81=E5=AE=8C=E6=88=90=E3=80=82?= =?UTF-8?q?=20update-eto:=20=20-=20rename:=20TrackPushEto=20->=20TrackPush?= =?UTF-8?q?CompletedEvent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Events/TrackPushCompletedEvent.cs | 82 +++++++++++ .../Events/TrackPushEvent.cs | 5 + .../EventHandlers/TrackPushEventHandler.cs | 137 +----------------- .../EventHandlers/TrackUpsertEventHandler.cs | 65 ++------- .../Rabbit/TrackPushCompletedEventHandler.cs | 103 +------------ .../Broadcasting/BroadcastingService.cs | 103 +++++++++++++ .../TrackPushRecordIndex.cs | 10 ++ 7 files changed, 223 insertions(+), 282 deletions(-) create mode 100644 src/SJZY.Track.Application.Contract/Events/TrackPushCompletedEvent.cs create mode 100644 src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs diff --git a/src/SJZY.Track.Application.Contract/Events/TrackPushCompletedEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackPushCompletedEvent.cs new file mode 100644 index 00000000..6d8646ef --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Events/TrackPushCompletedEvent.cs @@ -0,0 +1,82 @@ +namespace SJZY.Track.Application.Contract.Events; + +/// +/// 轨迹推送广播模型 +/// +public class TrackPushCompletedEvent +{ + /// + /// 是否为手动推送 + /// + public bool IsManualPush { get; set; } + + /// + /// 推送时间 + /// + public DateTime PushTime { get; set; } + + /// + /// 操作人 + /// + public string? OperationMan { get; set; } + + /// + /// 订单号 + /// + public string OrderNo { get; set; } = null!; + + /// + /// 接口通请求Id + /// + public Guid RequestId { get; set; } + + /// + /// 信息 + /// + public string Message { get; set; } = null!; + + /// + /// 操作类型 + /// + public string OptionType { get; set; } = null!; + + /// + /// 轨迹状态 + /// + public string TrackStatus { get; set; } = null!; + + /// + /// 轨迹内容代码 + /// + public string TrackCode { get; set; } = null!; + + /// + /// 轨迹内容 + /// + public string TrackContent { get; set; } = null!; + + /// + /// 轨迹发生地 + /// + public string TrackPlace { get; set; } = null!; + + /// + /// 轨迹发生时间 + /// + public DateTime TrackTime { get; set; } + + /// + /// 轨迹发生时区 + /// + public float TrackTimeZone { get; set; } + + /// + /// 是否成功 + /// + public bool IsSuccess { get; set; } + + /// + /// 推送内容（json） + /// + public string PushContent { get; set; } = null!; +} diff --git a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs index f51929db..ab033510 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs @@ -46,6 +46,11 @@ public class TrackPushEvent /// 是否强制推送 /// public bool ForcedPush { get; set; } + + /// + /// 是否发送到mq + /// + public bool IsToMq { get; set; } } public struct SystemPushStruct diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index bcec6d49..a4e74aeb 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -1,11 +1,8 @@ -using System.Text.Json; using DotNetCore.CAP; using Microsoft.Extensions.Options; using Newtonsoft.Json; -using Newtonsoft.Json.Linq; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; -using SJZY.OTel; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; @@ -127,6 +124,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CreateUser = GlobalConst.SystemUser, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq }; var item = await BuiltPushItem(config, log, @event.SystemPushInfo.Extra); item.Log = JsonConvert.SerializeObject(log); @@ -173,6 +172,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe TrackTime = track.TrackTime, TrackTimeZone = track.TrackTimeZone, CreateUser = @event.OperationMan, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq }; var item = await BuiltPushItem(config, log, track.Extra); item.Log = JsonConvert.SerializeObject(log); @@ -182,20 +183,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (items.Count == 0) return; - //获取参数结构名称 - // var paramNames = await GetExtraParamNamesAsync(@event.SubSystemCode); - // - // //如果为系统推送，则从事件中取出轨迹扩展内容填充 - // foreach (var item in items) - // { - // //根据事件中轨迹扩展内容填充 - // if (isSystemPush) - // item.Data.ForEach(dict => FillExtraParam(@event.SystemPushInfo.Extra, dict, paramNames)); - // //根据查出的轨迹中扩展内容填充 - // else - // item.Data.ForEach(dict => FillExtraParam(trackPushInputDict[item].Extra, dict, paramNames)); - // } - using (Track2InterfaceRelaySubSystemContext.Use( new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode))) { @@ -235,6 +222,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CreateUser = GlobalConst.SystemUser, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq }; var item = BuiltUpdateItem(config, log, @event.SystemPushInfo.Extra); item.Log = JsonConvert.SerializeObject(log); @@ -243,13 +232,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (items.Count == 0) return; - // //获取参数结构名称 - // var paramNames = await GetExtraParamNamesAsync(@event.SubSystemCode); - // - // //根据事件中轨迹扩展内容填充请求 - // foreach (var item in items) - // item.Data.ForEach(dict => FillExtraParam(@event.SystemPushInfo.Extra, dict, paramNames)); - using (Track2InterfaceRelaySubSystemContext.Use( new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode))) { @@ -384,114 +366,9 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { return obj == null ? string.Empty : System.Text.Json.JsonSerializer.Serialize(obj); } + private T? DeserializeForTextJson(string? json) { return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize(json); } - - - /* - * review by xiaolipro at 2024-12-24 - * todo：无用的方法，可以删除 @王云龙 - */ - - /// - /// 获取来源系统接口的参数名 - /// - /// - /// - private async Task> GetExtraParamNamesAsync(string subSystemCode) - { - using var span = ActivityHelper.CreateSpanActivity("开始请求接口通，接口结构"); - using (Track2InterfaceRelaySubSystemContext.Use( - new Track2InterfaceRelaySubSystemContext(subSystemCode))) - { - var res = await _httpClientProxy.Service.GetSystemInterfaceByIdAsync(_options.CurrentValue.PushTrackInterfaceCode, _options.CurrentValue.PushTrackInterfaceId); - if (res == null) return new List(); - span.SetTag("request-GetSystemInterfaceByIdAsync", $"PushTrackInterfaceCode[{subSystemCode}]," + - $"PushTrackInterfaceCode:[{_options.CurrentValue.PushTrackInterfaceCode}]," + - $"PushTrackInterfaceId:[{_options.CurrentValue.PushTrackInterfaceId}]," + - $"PushTrackItemParamAliasPrefix:[{_options.CurrentValue.PushTrackItemParamAliasPrefix}]"); - span.SetTag("response-GetSystemInterfaceByIdAsync", res.ToJson()); - return res.InParameterList - .Where(x => x.Alias.StartsWith(_options.CurrentValue.PushTrackItemParamAliasPrefix)) - .Select(x => x.Name) - .ToList(); - } - } - - /// - /// 填充来源于轨迹扩展字段的参数 - /// - /// 轨迹明细扩展字段Json - /// 要扩展的参数字典 - /// 来源扩展字段的参数名 - /// - private void FillExtraParam(string? trackExtraJson, Dictionary? dict, List? paramNames) - { - if (paramNames == null || paramNames.Count == 0) return; - if (dict == null) return; - - if (string.IsNullOrWhiteSpace(trackExtraJson)) return; - - dict.TryAdd("extra",JObject.FromObject(trackExtraJson)); - var trackExtra = ExtractTrackExtraDict(trackExtraJson); - - if (trackExtra.Count == 0) return; - foreach (var item in - trackExtra.Where(x => x.Value != null && paramNames.Contains(x.Key))) - { - dict.TryAdd(item.Key, item.Value); - } - } - - private Dictionary ExtractTrackExtraDict(string trackExtraStr) - { - var dict = new Dictionary(); - var trackExtra = JsonConvert.DeserializeObject>(trackExtraStr); - if (trackExtra == null) return dict; - foreach (var pair in trackExtra) - { - if (pair.Value is JsonElement jEl) - { - object? objVal = null; - switch (jEl.ValueKind) - { - case JsonValueKind.String: - objVal = jEl.GetString(); - // 处理字符串值 - break; - case JsonValueKind.Number: - objVal = jEl.GetDouble(); - // 处理数值 - break; - case JsonValueKind.True: - case JsonValueKind.False: - objVal = jEl.GetBoolean(); - // 处理布尔值 - break; - case JsonValueKind.Object: - // 处理对象 - break; - case JsonValueKind.Array: - // 处理数组 - break; - case JsonValueKind.Null: - // 处理空值 - break; - default: - // 处理未知类型 - break; - } - - if (objVal == null) continue; - dict.TryAdd(pair.Key, objVal); - continue; - } - - dict.TryAdd(pair.Key, pair.Value); - } - - return dict; - } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index c168aa11..c9fa94d1 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -1,9 +1,8 @@ -using System.Text; -using BSI.Stack.Rabbit; using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; +using SJZY.Track.Application.Services.Broadcasting; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Dtos.TrackContents; @@ -20,7 +19,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly RabbitClientBus _rabbitClientBus; + private readonly BroadcastingService _broadcastingService; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; public TrackUpsertEventHandler( @@ -28,15 +27,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository, - RabbitClientBus rabbitClientBus, - IOrderOperationRecordEsRepository orderOperationRecordEsRepository) + IOrderOperationRecordEsRepository orderOperationRecordEsRepository, + BroadcastingService broadcastingService) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; - _rabbitClientBus = rabbitClientBus; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; + _broadcastingService = broadcastingService; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -49,62 +48,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var content = await HandleTrackAsync(@event, canNextOrderIds); - // todo: 广播提取到一个服务里去，异步处理，不影响主流程 // 处理广播事件 - PublicTrackTopic(@event, content); + if (content is { IsToMq: true }) + await _broadcastingService.TrackChangePublicTopicAsync(@event, content); if (canNextOrderIds.Count == 0) return; // 根据订单关联关系异步向下刷（异步） await DownByOrderLink(@event, canNextOrderIds); } - /// - /// 发送轨迹广播 - /// - /// - /// - private void PublicTrackTopic(TrackUpsertEvent eto, TrackContentVO? content) - { - // var trackStatus = await _trackStatusRepository.GetFirstAsync(m => m.Code == content.TrackStatusCode); - if (content == null || !content.IsToMq) - return; - - var data = new TrackChangeEto() - { - EstimatedDeliveryDate = eto.EstimatedDeliveryDate, - TrackPlacePostcode = eto.TrackPlacePostcode, - TrackPlaceNation = eto.TrackPlaceNation, - TrackPlaceNationId = eto.TrackPlaceNationId, - TrackPlaceProvince = eto.TrackPlaceProvince, - TrackPlaceProvinceId = eto.TrackPlaceProvinceId, - TrackPlaceCity = eto.TrackPlaceCity, - TrackPlaceCityId = eto.TrackPlaceCityId, - Files = eto.Files, - Remarks = eto.Remarks, - Source = eto.Source, - IsVisibility = eto.IsVisibility, - TrackTimeZone = eto.TrackTimeZone, - TrackTime = eto.TrackTime, - TrackPlace = eto.TrackPlace, - TrackStatusCode = content.TrackStatusCode, - EnStatus = eto.EnStatus, - Content = eto.Content, - TrackCode = eto.TrackCode, - OperationMan = eto.OperationMan, - OrderIdentifiers = eto.OrderIdentifiers, - CustomId = eto.CustomId, - SubSystemId = eto.SubSystemId, - Extra = eto.Extra - }; - - if (eto.IsAdd) - data.Option = 1; - else - data.Option = 3; - - _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); - } - private async Task DownByOrderLink(TrackUpsertEvent @event, List canNextOrderIds) { // 根据订单关联关系异步向下刷（异步） @@ -268,7 +220,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackTimeZone = @event.TrackTimeZone, Extra = @event.Extra }, - ForcedPush = @event.ForcedPush + ForcedPush = @event.ForcedPush, + IsToMq = trackContent.IsToMq }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); diff --git a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs index 2e2c8c56..3a5fed9f 100644 --- a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -4,17 +4,12 @@ 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.Application.Services.Broadcasting; 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; @@ -24,24 +19,15 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl { private readonly IOrderRepository _orderRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; - private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; - private readonly ITrackStatusRepository _trackStatusRepository; - private readonly ITrackContentRepository _trackContentRepository; - private readonly RabbitClientBus _rabbitClientBus; + private readonly BroadcastingService _broadcastingService; public TrackPushCompletedEventHandler(IOrderRepository orderRepository, ITrackPushRecordEsRepository trackPushRecordEsRepository, - ITrackDetailsEsRepository trackDetailsEsRepository, - ITrackStatusRepository trackStatusRepository, - ITrackContentRepository trackContentRepository, - RabbitClientBus rabbitClientBus) + BroadcastingService broadcastingService) { _orderRepository = orderRepository; _trackPushRecordEsRepository = trackPushRecordEsRepository; - _trackDetailsEsRepository = trackDetailsEsRepository; - _trackStatusRepository = trackStatusRepository; - _trackContentRepository = trackContentRepository; - _rabbitClientBus = rabbitClientBus; + _broadcastingService = broadcastingService; } public ConsumeOptions Options() @@ -71,7 +57,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl } else if (optionType == OptionType.修改.ToString()) { - logStr = JsonConvert.DeserializeObject>(@event.RequestParams)!.First().Log;; + logStr = JsonConvert.DeserializeObject>(@event.RequestParams)!.First().Log; } if (logStr.IsNullOrEmpty()) throw new InvalidOperationException(); @@ -82,16 +68,9 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl 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); + if (log.IsToMq) + await _broadcastingService.TrackPushCompletedPublicTopicAsync(log); // 往es 写推送日志 await _trackPushRecordEsRepository.AddAsync(log); @@ -140,72 +119,4 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl }); } } - - /// - /// 轨迹推送广播 - /// - 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())); - } } diff --git a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs new file mode 100644 index 00000000..68b3ddea --- /dev/null +++ b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs @@ -0,0 +1,103 @@ +using System.Text; +using BSI.Stack.Rabbit; +using SJZY.Expand.ABP.Core.PublicMethords; +using SJZY.Track.Application.Contract.Events; +using SJZY.Track.Domain.Aggregates.TrackPushAggregate; +using SJZY.Track.Domain.Dtos.TrackContents; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.Services.Broadcasting; + +/// +/// 广播服务 +/// +public class BroadcastingService : ITransientDependency +{ + private readonly RabbitClientBus _rabbitClientBus; + + public BroadcastingService(RabbitClientBus rabbitClientBus) + { + _rabbitClientBus = rabbitClientBus; + } + + /// + /// 轨迹变更广播 + /// + /// + /// + /// + public Task TrackChangePublicTopicAsync(TrackUpsertEvent eto, TrackContentVO content) + { + return Task.Run(() => + { + var data = new TrackChangeEto + { + EstimatedDeliveryDate = eto.EstimatedDeliveryDate, + TrackPlacePostcode = eto.TrackPlacePostcode, + TrackPlaceNation = eto.TrackPlaceNation, + TrackPlaceNationId = eto.TrackPlaceNationId, + TrackPlaceProvince = eto.TrackPlaceProvince, + TrackPlaceProvinceId = eto.TrackPlaceProvinceId, + TrackPlaceCity = eto.TrackPlaceCity, + TrackPlaceCityId = eto.TrackPlaceCityId, + Files = eto.Files, + Remarks = eto.Remarks, + Source = eto.Source, + IsVisibility = eto.IsVisibility, + TrackTimeZone = eto.TrackTimeZone, + TrackTime = eto.TrackTime, + TrackPlace = eto.TrackPlace, + TrackStatusCode = content.TrackStatusCode, + EnStatus = eto.EnStatus, + Content = eto.Content, + TrackCode = eto.TrackCode, + OperationMan = eto.OperationMan, + OrderIdentifiers = eto.OrderIdentifiers, + CustomId = eto.CustomId, + SubSystemId = eto.SubSystemId, + Extra = eto.Extra + }; + + if (eto.IsAdd) + data.Option = 1; + else + data.Option = 3; + + _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); + return Task.CompletedTask; + }); + } + + /// + /// 轨迹推送完成广播 + /// + /// + /// + public Task TrackPushCompletedPublicTopicAsync(TrackPushRecordIndex log) + { + return Task.Run(() => + { + var data = new TrackPushCompletedEvent + { + 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 + }; + + _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); + return Task.CompletedTask; + }); + } +} diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs index f983a5a6..c8731039 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs @@ -82,4 +82,14 @@ public class TrackPushRecordIndex : IdBaseIndex /// 创建时间 public DateTime CreateTime { get; set; } + + /// + /// 子系统代码 + /// + public string SubSystemCode { get; set; } + + /// + /// 是否广播 + /// + public bool IsToMq { get; set; } }