liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- elm self delivery support.

Browse Source
This commit is contained in:
gazebo
2018-07-28 11:49:54 +08:00
parent ad93408e4c
commit 7734399392
4 changed files with 154 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
business/scheduler/defsch/defsch.go
View File

@@ -32,6 +32,10 @@ const (
)
type WatchOrderInfo struct {
isNeedAutoPickup bool
storeDeliveryType int
isNeed3rdDelivery bool
order *model.GoodsOrder // order里的信息是保持更新的
waybills []*model.Waybill // 这个waybills里的状态信息是不真实的，只使用id相关的信息
@@ -42,24 +46,81 @@ type WatchOrderInfo struct {
retryCount int // 失败后尝试的次数，调试阶段可能出现死循化，阻止这种情况发生
}
type StatusActionConfig struct {
TimerType int // 参见上面的相关常量定义
Timeout time.Duration // 超时时间，0在GetStatusActionConfig返回时表示不修改缺省
TimeoutGap int // 以秒为单位的随机时间，0在GetStatusActionConfig返回时表示不修改缺省
TimeoutAction func(savedOrderInfo *WatchOrderInfo) (err error) // 超时后需要执行的动作，为nil表示此状态不需要执行监控， nil在GetStatusActionConfig返回时表示不修改缺省
}
// 重要：此调度器要求同一定单的处理逻辑必须是序列化了的，不然会有并发问题
type DefScheduler struct {
scheduler.BaseScheduler
defWorkflowConfig []map[int]*scheduler.StatusActionConfig
defWorkflowConfig []map[int]*StatusActionConfig
orderMap jxutils.SyncMapWithTimeout
}
func NewWatchOrderInfo(order *model.GoodsOrder) (retVal *WatchOrderInfo) {
retVal = &WatchOrderInfo{
isNeedAutoPickup: true,
isNeed3rdDelivery: false,
storeDeliveryType: scheduler.StoreDeliveryTypeCrowdSourcing,
}
retVal.SetOrder(order)
return retVal
}
func (s *WatchOrderInfo) SetOrder(order *model.GoodsOrder) (retVal *model.GoodsOrder) {
retVal = s.order
if s.order != order {
if order != nil {
s.updateOrderStoreFeature(order)
}
s.order = order
}
return retVal
}
func (s *WatchOrderInfo) updateOrderStoreFeature(order *model.GoodsOrder) (err error) {
storefeature := &model.Jxstorefeature{
Id: jxutils.GetJxStoreIDFromOrder(order),
}
if storefeature.Id > 0 {
db := orm.NewOrm()
utils.CallFuncLogError(func() error {
err = db.Read(storefeature, "Id")
if err == nil {
s.isNeedAutoPickup = (storefeature.Autopickup != 0)
if order.VendorID == model.VendorIDELM {
s.storeDeliveryType = int(storefeature.ElmDeliveryType)
}
if s.storeDeliveryType != scheduler.StoreDeliveryTypeByStore {
if (order.VendorID == model.VendorIDJD && storefeature.Deliverycompetition != 0) ||
(order.VendorID == model.VendorIDELM && storefeature.Transmtzs != 0) {
s.isNeed3rdDelivery = true
}
} else {
s.isNeed3rdDelivery = true
}
}
return err
}, "updateOrderStoreFeature")
}
return err
}
func init() {
sch := &DefScheduler{}
sch.IsReallyCallPlatformAPI = globals.ReallyCallPlatformAPI
sch.Init()
scheduler.CurrentScheduler = sch
sch.defWorkflowConfig = []map[int]*scheduler.StatusActionConfig{
map[int]*scheduler.StatusActionConfig{
model.OrderStatusNew: &scheduler.StatusActionConfig{ // 自动接单
sch.defWorkflowConfig = []map[int]*StatusActionConfig{
map[int]*StatusActionConfig{
model.OrderStatusNew: &StatusActionConfig{ // 自动接单
TimerType: scheduler.TimerTypeBaseNow,
Timeout: 1 * time.Second,
TimeoutAction: func(order *model.GoodsOrder) (err error) {
TimeoutAction: func(savedOrderInfo *WatchOrderInfo) (err error) {
order := savedOrderInfo.order
_ = sch.handleAutoAcceptOrder(order.VendorOrderID, order.VendorID, order.ConsigneeMobile, jxutils.GetJxStoreIDFromOrder(order), nil, func(isAcceptIt bool) error {
if err = sch.AcceptOrRefuseOrder(order, isAcceptIt); err != nil {
// 为了解决京东新消息与接单消息乱序的问题
@@ -77,21 +138,38 @@ func init() {
return nil
},
},
model.OrderStatusAccepted: &scheduler.StatusActionConfig{ // 自动拣货
model.OrderStatusAccepted: &StatusActionConfig{ // 自动拣货
TimerType: scheduler.TimerTypeBaseExpectedDeliveredTime,
Timeout: time2AutoPickupMin,
TimeoutGap: time2AutoPickupGap,
TimeoutAction: func(order *model.GoodsOrder) (err error) {
return sch.autoPickupGood(order)
TimeoutAction: func(savedOrderInfo *WatchOrderInfo) (err error) {
if savedOrderInfo.isNeedAutoPickup {
return sch.autoPickupGood(savedOrderInfo.order)
}
return nil
},
},
model.OrderStatusFinishedPickup: &StatusActionConfig{
TimerType: scheduler.TimerTypeBaseNow,
Timeout: 1 * time.Second,
TimeoutGap: 0,
TimeoutAction: func(savedOrderInfo *WatchOrderInfo) (err error) {
if savedOrderInfo.storeDeliveryType == scheduler.StoreDeliveryTypeByStore { // 自配置商家使用
return sch.createWaybillOn3rdProviders(savedOrderInfo, nil)
}
return nil
},
},
},
map[int]*scheduler.StatusActionConfig{
model.WaybillStatusNew: &scheduler.StatusActionConfig{ // 尝试召唤更多物流
map[int]*StatusActionConfig{
model.WaybillStatusNew: &StatusActionConfig{
TimerType: scheduler.TimerTypeBaseStatusTime,
Timeout: time2Schedule3rdCarrier,
TimeoutAction: func(order *model.GoodsOrder) (err error) {
return sch.createWaybillOn3rdProviders(order, nil)
TimeoutAction: func(savedOrderInfo *WatchOrderInfo) (err error) {
if savedOrderInfo.storeDeliveryType != scheduler.StoreDeliveryTypeByStore { // 非自配置商家使用
return sch.createWaybillOn3rdProviders(savedOrderInfo, nil)
}
return nil
},
},
},
@@ -103,12 +181,10 @@ func (s *DefScheduler) OnOrderNew(order *model.GoodsOrder, isPending bool) (err
globals.SugarLogger.Debugf("OnOrderNew orderID:%s", order.VendorOrderID)
savedOrderInfo := s.loadSavedOrderFromMap(model.Order2Status(order), false)
if savedOrderInfo == nil {
savedOrderInfo = &WatchOrderInfo{
order: order,
}
savedOrderInfo = NewWatchOrderInfo(order)
s.orderMap.StoreWithTimeout(jxutils.ComposeUniversalOrderID(order.VendorOrderID, order.VendorID), savedOrderInfo, orderMapStoreMaxTime)
} else {
savedOrderInfo.order = order // 调整单或消息错序都可能进到这里来
savedOrderInfo.SetOrder(order) // 调整单或消息错序都可能进到这里来
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeOrder, savedOrderInfo.order.Status, false)
return err
@@ -147,9 +223,9 @@ func (s *DefScheduler) OnWaybillStatusChanged(bill *model.Waybill, isPending boo
if !isPending {
if order.WaybillVendorID != model.VendorIDUnknown {
globals.SugarLogger.Debugf("OnWaybillStatusChanged multiple waybill created, bill:%v", bill)
if !(order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID) && bill.WaybillVendorID != order.VendorID {
if !s.isBillCandidate(order, bill) && bill.WaybillVendorID != order.VendorID {
s.CancelWaybill(bill)
} else if bill.WaybillVendorID == order.VendorID {
} else if bill.WaybillVendorID == order.VendorID && order.WaybillVendorID != order.VendorID {
globals.SugarLogger.Warnf("OnWaybillStatusChanged bill:%v purchase platform bill came later than others, strange!!!", bill)
}
}
@@ -161,61 +237,69 @@ func (s *DefScheduler) OnWaybillStatusChanged(bill *model.Waybill, isPending boo
} else {
switch bill.Status {
case model.WaybillStatusAccepted:
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if !isPending {
if order.WaybillVendorID == model.VendorIDUnknown || bill.WaybillVendorID == order.VendorID { // 购买平台的运单，优先级最高
s.updateOrderByBill(order, bill, false)
s.cancelOtherWaybills(savedOrderInfo, bill)
} else if !(order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID) && bill.WaybillVendorID != order.VendorID {
if bill.WaybillVendorID != bill.OrderVendorID {
if savedOrderInfo.storeDeliveryType == scheduler.StoreDeliveryTypeByStore {
s.SelfDeliverDelievering(savedOrderInfo.order)
} else {
s.swtich2SelfDeliverWithRetry(savedOrderInfo.order, bill, 2, 10*time.Second)
}
}
} else if !s.isBillCandidate(order, bill) && bill.WaybillVendorID != order.VendorID {
// todo 当前逻辑，加载PENDING的ORDER时，正常状态也可能进这里
s.CancelWaybill(bill)
globals.SugarLogger.Warnf("OnWaybillStatusChanged Accepted orderID:%s got multiple bill:%v, order details:%v", order.VendorOrderID, bill, order)
}
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
case model.WaybillStatusAcceptCanceled:
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
if s.isBillCandidate(order, bill) {
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if !isPending {
bill.WaybillVendorID = model.VendorIDUnknown
s.updateOrderByBill(order, bill, false)
s.createWaybillOn3rdProviders(order, bill)
s.createWaybillOn3rdProviders(savedOrderInfo, bill)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
} else if order.WaybillVendorID != model.VendorIDUnknown {
s.CancelWaybill(bill)
globals.SugarLogger.Warnf("OnWaybillStatusChanged AcceptCanceled orderID:%s got multiple bill:%v, order details:%v", order.VendorOrderID, bill, order)
}
case model.WaybillStatusCourierArrived: // do nothing
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if s.isBillCandidate(order, bill) {
} else {
// s.CancelWaybill(bill)
globals.SugarLogger.Warnf("OnWaybillStatusChanged CourierArrived order(%d, %s) bill(%d, %s), bill:%v shouldn't got here", order.WaybillVendorID, order.VendorWaybillID, bill.WaybillVendorID, bill.VendorWaybillID, bill)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
case model.WaybillStatusFailed: // todo WaybillStatusFailed理解成订单整个失败了，不需要再尝试创建运单了，注意这里应该加个zabbix日志的报警
s.removeWaybillFromMap(savedOrderInfo, bill)
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
if s.isBillCandidate(order, bill) {
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if !isPending {
globals.SugarLogger.Infof("OnWaybillStatusChanged WaybillStatusFailed, bill:%v", bill)
bill.WaybillVendorID = model.VendorIDUnknown
s.updateOrderByBill(order, bill, true)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
} else {
globals.SugarLogger.Warnf("OnWaybillStatusChanged Failed bill:%v shouldn't got here, order details:%v", bill, order)
}
case model.WaybillStatusCanceled:
s.removeWaybillFromMap(savedOrderInfo, bill)
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
if s.isBillCandidate(order, bill) {
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if !isPending {
bill.WaybillVendorID = model.VendorIDUnknown
s.updateOrderByBill(order, bill, true)
s.createWaybillOn3rdProviders(order, nil)
s.createWaybillOn3rdProviders(savedOrderInfo, nil)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
}
case model.WaybillStatusDelivering:
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
if s.isBillCandidate(order, bill) {
if order.VendorID != bill.WaybillVendorID && !isPending {
s.SelfDeliverDelievering(order)
}
@@ -223,17 +307,16 @@ func (s *DefScheduler) OnWaybillStatusChanged(bill *model.Waybill, isPending boo
// s.CancelWaybill(bill)
globals.SugarLogger.Warnf("OnWaybillStatusChanged Delivering order(%d, %s) bill(%d, %s), bill:%v shouldn't got here", order.WaybillVendorID, order.VendorWaybillID, bill.WaybillVendorID, bill.VendorWaybillID, bill)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
case model.WaybillStatusDelivered:
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
s.removeWaybillFromMap(savedOrderInfo, bill)
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
if s.isBillCandidate(order, bill) {
if order.VendorID != bill.WaybillVendorID && !isPending {
s.SelfDeliverDelievered(order)
}
} else {
globals.SugarLogger.Warnf("OnWaybillStatusChanged Delivered order(%d, %s) bill(%d, %s), bill:%v shouldn't got here", order.WaybillVendorID, order.VendorWaybillID, bill.WaybillVendorID, bill.VendorWaybillID, bill)
}
s.resetTimer(savedOrderInfo, scheduler.TimerStatusTypeWaybill, bill.Status, false)
}
}
}
@@ -261,13 +344,13 @@ func (s *DefScheduler) removeWaybillFromMap(savedOrderInfo *WatchOrderInfo, bill
}
}
func (s *DefScheduler) createWaybillOn3rdProviders(order *model.GoodsOrder, excludeBill *model.Waybill) (err error) {
func (s *DefScheduler) createWaybillOn3rdProviders(savedOrderInfo *WatchOrderInfo, excludeBill *model.Waybill) (err error) {
order := savedOrderInfo.order
globals.SugarLogger.Debugf("createWaybillOn3rdProviders, orderID:%s, status:%d, excludeBill:%v", order.VendorOrderID, order.Status, excludeBill)
savedOrderInfo := s.loadSavedOrderFromMap(model.Order2Status(order), true)
savedOrderInfo.retryCount++
if savedOrderInfo.retryCount <= maxWaybillRetryCount {
if order.Status == model.OrderStatusFinishedPickup {
if s.isOrderSupport3rdDelivery(order) {
if savedOrderInfo.isNeed3rdDelivery || savedOrderInfo.storeDeliveryType == scheduler.StoreDeliveryTypeByStore {
successCount := 0
for vendorID := range s.DeliveryPlatformHandlers {
if (excludeBill == nil || vendorID != excludeBill.WaybillVendorID) && s.DeliveryPlatformHandlers[vendorID].Use4CreateWaybill {
@@ -299,9 +382,6 @@ func (s *DefScheduler) cancelOtherWaybills(savedOrderInfo *WatchOrderInfo, bill2
s.CancelWaybill(v)
}
}
if bill2Keep != nil && bill2Keep.WaybillVendorID != bill2Keep.OrderVendorID {
s.swtich2SelfDeliverWithRetry(savedOrderInfo.order, bill2Keep, 2, 10*time.Second)
}
return nil
}
@@ -312,7 +392,7 @@ func (s *DefScheduler) swtich2SelfDeliverWithRetry(order *model.GoodsOrder, bill
globals.SugarLogger.Infof("swtich2SelfDeliverWithRetry failed, cancel bill:%v, err:%v", bill, err)
if s.CancelWaybill(bill) == nil {
// 转自送失败的取消，要将订单中的运单状态更新
if order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID {
if s.isBillCandidate(order, bill) {
bill.WaybillVendorID = model.VendorIDUnknown
s.updateOrderByBill(order, bill, false)
}
@@ -330,22 +410,22 @@ func (s *DefScheduler) loadSavedOrderFromMap(status *model.OrderStatus, isAutoLo
}
if isAutoLoad && (realSavedInfo == nil || !model.IsOrderSolid(realSavedInfo.order)) {
if realSavedInfo == nil {
realSavedInfo = new(WatchOrderInfo)
realSavedInfo = NewWatchOrderInfo(nil)
s.orderMap.StoreWithTimeout(universalOrderID, realSavedInfo, orderMapStoreMaxTime)
} else {
globals.SugarLogger.Infof("loadSavedOrderFromMap order is incomplete, orderID:%s, load it", status.RefVendorOrderID)
}
if order, err := s.CurOrderManager.LoadOrder(status.RefVendorOrderID, status.RefVendorID); err == nil {
realSavedInfo.order = order
realSavedInfo.SetOrder(order)
} else {
realSavedInfo.order = &model.GoodsOrder{
realSavedInfo.SetOrder(&model.GoodsOrder{
VendorOrderID: status.RefVendorOrderID,
VendorID: status.RefVendorID,
Status: status.Status,
StatusTime: status.StatusTime,
OrderCreatedAt: status.StatusTime,
WaybillVendorID: model.VendorIDUnknown,
}
})
globals.SugarLogger.Infof("loadSavedOrderFromMap can not load order orderID:%s", status.VendorOrderID)
}
}
@@ -365,9 +445,8 @@ func (s *DefScheduler) stopTimer(savedOrderInfo *WatchOrderInfo) {
func (s *DefScheduler) resetTimer(savedOrderInfo *WatchOrderInfo, statusType, status int, isPending bool) {
order := savedOrderInfo.order
globals.SugarLogger.Debugf("resetTimer, orderID:%s status:%v", order.VendorOrderID, status)
if statusType != savedOrderInfo.timerStatusType || status >= savedOrderInfo.timerStatus { // 新设置的TIMER不能覆盖状态在其后的TIMER，如果状态回绕，需要注意
config := s.mergeOrderStatusConfig(statusType, status, s.GetPurchasePlatformFromVendorID(order.VendorID).GetStatusActionConfig(statusType, status))
config := s.mergeOrderStatusConfig(statusType, status, order.VendorID)
if config == nil || config.TimerType != scheduler.TimerTypeByPass {
s.stopTimer(savedOrderInfo)
}
@@ -396,7 +475,7 @@ func (s *DefScheduler) resetTimer(savedOrderInfo *WatchOrderInfo, statusType, st
timeout = 0
}
if timeout == 0 {
config.TimeoutAction(order)
config.TimeoutAction(savedOrderInfo)
} else {
timerName := ""
if statusType == scheduler.TimerStatusTypeOrder {
@@ -409,7 +488,7 @@ func (s *DefScheduler) resetTimer(savedOrderInfo *WatchOrderInfo, statusType, st
savedOrderInfo.timer = time.AfterFunc(timeout, func() {
jxutils.CallMsgHandlerAsync(func() {
globals.SugarLogger.Debugf("fire timer:%s, orderID:%s", timerName, order.VendorOrderID)
config.TimeoutAction(order)
config.TimeoutAction(savedOrderInfo)
savedOrderInfo.timerStatus = 0
savedOrderInfo.timerStatusType = scheduler.TimerStatusTypeUnknown
}, order.VendorOrderID)
@@ -452,28 +531,18 @@ func (s *DefScheduler) handleAutoAcceptOrder(orderID string, vendorID int, userM
return handleType
}
func (s *DefScheduler) mergeOrderStatusConfig(statusType, status int, config *scheduler.StatusActionConfig) (retVal *scheduler.StatusActionConfig) {
func (s *DefScheduler) mergeOrderStatusConfig(statusType, status int, purchaseVendorID int) (retVal *StatusActionConfig) {
vendorTimeout := s.GetPurchasePlatformFromVendorID(purchaseVendorID).GetStatusActionTimeout(statusType, status)
defConfig := s.defWorkflowConfig[statusType][status]
if defConfig == nil && config == nil {
if defConfig == nil {
return nil
}
retVal = &scheduler.StatusActionConfig{}
retVal = &StatusActionConfig{}
if defConfig != nil {
*retVal = *defConfig
}
if config != nil {
if config.Timeout != 0 {
retVal.Timeout = config.Timeout
}
if config.TimerType != scheduler.TimerTypeNoOverride {
retVal.TimerType = config.TimerType
}
if config.TimeoutGap != 0 {
retVal.TimeoutGap = config.TimeoutGap
}
if config.TimeoutAction != nil {
retVal.TimeoutAction = config.TimeoutAction
}
if vendorTimeout != 0 {
retVal.Timeout = vendorTimeout
}
return retVal
}
@@ -485,43 +554,6 @@ func (s *DefScheduler) updateOrderByStatus(order *model.GoodsOrder, status *mode
return order
}
func (s *DefScheduler) isOrderSupport3rdDelivery(order *model.GoodsOrder) (retVal bool) {
storefeature := &model.Jxstorefeature{
Id: jxutils.GetJxStoreIDFromOrder(order),
}
db := orm.NewOrm()
utils.CallFuncLogError(func() error {
err := db.Read(storefeature, "Id")
if err == nil {
if (order.VendorID == model.VendorIDJD && storefeature.Deliverycompetition == 1) ||
(order.VendorID == model.VendorIDELM && storefeature.Transmtzs == 1) {
retVal = true
}
}
return err
}, "isOrderSupport3rdDelivery")
return retVal
}
func (s *DefScheduler) isOrderSupportAutoPickup(order *model.GoodsOrder) (retVal bool) {
retVal = true
storefeature := &model.Jxstorefeature{
Id: jxutils.GetJxStoreIDFromOrder(order),
}
db := orm.NewOrm()
utils.CallFuncLogError(func() error {
err := db.Read(storefeature, "Id")
if err == nil || err == orm.ErrNoRows {
if storefeature.Autopickup == 0 {
retVal = false
}
err = nil
}
return err
}, "isOrderSupportAutoPickup")
return retVal
}
func (s *DefScheduler) updateOrderByBill(order *model.GoodsOrder, bill *model.Waybill, revertStatus bool) {
if bill.WaybillVendorID == model.VendorIDUnknown {
bill.VendorWaybillID = ""
@@ -535,10 +567,9 @@ func (s *DefScheduler) updateOrderByBill(order *model.GoodsOrder, bill *model.Wa
}
func (s *DefScheduler) autoPickupGood(order *model.GoodsOrder) (err error) {
if s.isOrderSupportAutoPickup(order) {
err = s.PickedUpGoods(order)
} else {
globals.SugarLogger.Debugf("autoPickupGood orderID:%s doesn't support auto pickup", order.VendorOrderID)
}
return err
return s.PickedUpGoods(order)
}
func (s *DefScheduler) isBillCandidate(order *model.GoodsOrder, bill *model.Waybill) bool {
return order.WaybillVendorID == bill.WaybillVendorID && order.VendorWaybillID == bill.VendorWaybillID
}