Merge remote-tracking branch 'origin/S03-轨迹中台优化项' into test
This commit is contained in:
dingenwen
commit
1b6d426550
@ -5,7 +5,7 @@ public class TrackChangeAfterUpdateOrderEvent
|
||||
/// <summary>
|
||||
/// 订单Id
|
||||
/// </summary>
|
||||
public int[] OrderIds { get; set; }
|
||||
public int[] OrderIds { get; set; } = null!;
|
||||
|
||||
/// <summary>
|
||||
/// 是否删除引起的更新
|
||||
|
||||
@ -2,6 +2,11 @@
|
||||
|
||||
public class TrackPushEvent
|
||||
{
|
||||
/// <summary>
|
||||
/// 重试次数
|
||||
/// </summary>
|
||||
public int RetryNum { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 操作类型
|
||||
/// </summary>
|
||||
|
||||
@ -20,15 +20,17 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
|
||||
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
|
||||
private readonly IOrderRepository _orderRepository;
|
||||
private readonly ITrackPushCacheService _trackPushCacheService;
|
||||
private readonly ICapPublisher _capPublisher;
|
||||
|
||||
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
|
||||
ITrackPushRecordsRepository trackPushRecordsRepository,
|
||||
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService)
|
||||
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher)
|
||||
{
|
||||
_httpClientProxy = httpClientProxy;
|
||||
_trackPushRecordsRepository = trackPushRecordsRepository;
|
||||
_orderRepository = orderRepository;
|
||||
_trackPushCacheService = trackPushCacheService;
|
||||
_capPublisher = capPublisher;
|
||||
}
|
||||
|
||||
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
|
||||
@ -103,6 +105,19 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
|
||||
throw new ArgumentOutOfRangeException();
|
||||
}
|
||||
|
||||
if (!log.IsSuccess)
|
||||
{
|
||||
if (@event.RetryNum < 3)
|
||||
{
|
||||
// 1, 2, 4min
|
||||
await _capPublisher.PublishDelayAsync(
|
||||
TimeSpan.FromSeconds((1 << @event.RetryNum) * 60),
|
||||
GlobalConst.TrackPushSubscribe, @event);
|
||||
@event.RetryNum++;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// 往es 写推送日志
|
||||
await _trackPushRecordsRepository.AddAsync(log);
|
||||
|
||||
@ -169,7 +184,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
|
||||
Extra = config.ExpandFieldsValue
|
||||
};
|
||||
|
||||
var body = new TrackPushInput()
|
||||
var body = new TrackPushInput
|
||||
{
|
||||
OrderNo = @event.OrderNo,
|
||||
};
|
||||
|
||||
@ -105,7 +105,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
if (track != null && @event.IsAdd!.Value) return; // 存在不让创建
|
||||
}
|
||||
|
||||
// step1、upsert es轨迹
|
||||
if (track == default)
|
||||
{
|
||||
track = new TrackDetailsIndex
|
||||
@ -159,8 +158,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
var trackPushEvent = new TrackPushEvent
|
||||
{
|
||||
OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改,
|
||||
OrderId = track.OrderId,
|
||||
OrderNo = track.OrderNo,
|
||||
OrderId = order.OrderId,
|
||||
TrackStatus = track.TrackStatus,
|
||||
TrackCode = track.TrackCode,
|
||||
TrackContent = track.TrackContent,
|
||||
@ -174,6 +172,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
|
||||
var firstTrack = insertTracks.FirstOrDefault();
|
||||
if (firstTrack == null) return;
|
||||
|
||||
// step1、upsert es轨迹
|
||||
await _trackDetailsRepository.AddManyAsync(insertTracks);
|
||||
|
||||
// step3、触发订单主表更新事件（异步）
|
||||
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
|
||||
new TrackChangeAfterUpdateOrderEvent
|
||||
@ -187,7 +189,5 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
TrackTime = firstTrack.TrackTime,
|
||||
TrackTimeZone = firstTrack.TrackTimeZone
|
||||
});
|
||||
|
||||
await _trackDetailsRepository.AddManyAsync(insertTracks);
|
||||
}
|
||||
}
|
||||
@ -43,18 +43,16 @@ public class TrackPushService : ApplicationService
|
||||
{
|
||||
input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空
|
||||
|
||||
var orders = await _orderRepository.GetListAsync(x =>
|
||||
x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new
|
||||
{
|
||||
x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId
|
||||
});
|
||||
var orderIds = await _orderRepository.GetArrayAsync(x =>
|
||||
x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id);
|
||||
|
||||
if (orders.Count == 0)
|
||||
if (!orderIds.Any())
|
||||
{
|
||||
throw new BusinessException(message: "一个推送配置都不存在");
|
||||
throw new BusinessException(message: "一个订单都不存在");
|
||||
}
|
||||
|
||||
var tracks = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, input.TrackCode);
|
||||
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, input.TrackCode);
|
||||
tracks = tracks.DistinctBy(x => x.OrderId).ToList();
|
||||
|
||||
if (tracks.Count == 0)
|
||||
{
|
||||
|
||||
@ -19,8 +19,6 @@ public interface ITrackDetailsRepository
|
||||
|
||||
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos);
|
||||
|
||||
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode);
|
||||
|
||||
Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input);
|
||||
Task<List<TrackDetailsIndex>> GetListAsync(int orderId);
|
||||
|
||||
|
||||
@ -22,6 +22,9 @@ public interface IBasicQuery<T> where T : IEntity
|
||||
Task<List<TResult>> GetListAsync<TResult>(Expression<Func<T, bool>> whereExpression,
|
||||
Expression<Func<T, TResult>> resultExpression);
|
||||
|
||||
Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
|
||||
Expression<Func<T, TResult>> resultExpression);
|
||||
|
||||
/// <summary>
|
||||
/// count
|
||||
/// </summary>
|
||||
|
||||
@ -24,4 +24,10 @@ public class BasicQuery<T> : SimpleClient<T>, IBasicQuery<T> where T : class, IE
|
||||
{
|
||||
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToListAsync();
|
||||
}
|
||||
|
||||
public async Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
|
||||
Expression<Func<T, TResult>> resultExpression)
|
||||
{
|
||||
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToArrayAsync();
|
||||
}
|
||||
}
|
||||
@ -104,8 +104,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
|
||||
{
|
||||
if (!tracks.Any()) return;
|
||||
|
||||
// 按天写入es
|
||||
var r = await Context.IndexManyAsync(tracks, tracks.First().IndexName);
|
||||
|
||||
@ -141,33 +139,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
return data.Documents.ToList();
|
||||
}
|
||||
|
||||
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode)
|
||||
{
|
||||
if (orderNos.Count < 1) return new List<TrackDetailsIndex>();
|
||||
|
||||
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Source(s => s.Includes(fd => fd
|
||||
.Field(f => f.OrderId)
|
||||
.Field(f => f.OrderNo)
|
||||
.Field(f => f.TrackStatus)
|
||||
.Field(f => f.TrackCode)
|
||||
.Field(f => f.TrackContent)
|
||||
.Field(f => f.TrackPlace)
|
||||
.Field(f => f.TrackTime)
|
||||
.Field(f => f.TrackTimeZone)
|
||||
))
|
||||
.Query(q =>
|
||||
q.Terms(t => t
|
||||
.Field(f => f.OrderNo.Suffix("keyword"))
|
||||
.Terms(orderNos))
|
||||
&& q.Term(f => f.OrderType, orderType)
|
||||
&& q.Term(f => f.TrackCode.Suffix("keyword"), trackCode)
|
||||
)
|
||||
);
|
||||
|
||||
return data.Documents.ToList();
|
||||
}
|
||||
|
||||
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)
|
||||
{
|
||||
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
|
||||
Loading…
Reference in New Issue
Block a user