jx-callback/business/msghub/msghub.go

package msghub

import (
	"git.rosy.net.cn/jx-callback/business/jxstore/permission"
	"time"

	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/jx-callback/business/jxutils"
	"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
	"git.rosy.net.cn/jx-callback/business/model"
	"git.rosy.net.cn/jx-callback/business/model/dao"
	"git.rosy.net.cn/jx-callback/globals"
)

const (
	ServerMsgRegister   = "register"
	ServerMsgUnregister = "unregister"

	ServerMsgPing = "ping"

	ServerMsgNewOrder              = "newOrder"
	ServerMsgFinishedPickup        = "finishedPickup"        // 此消息当前当成刷新订单列表及状态来使用，并不是拣货
	ServerMsgKeyOrderStatusChanged = "keyOrderStatusChanged" // 重要订单状态变化

	ServerMsgNewWait4ApproveAfsOrder  = "newWait4ApproveAfsOrder"
	ServerMsgKeyAfsOrderStatusChanged = "keyAfsOrderStatusChanged" // 重要售后单状态变化
)

const (
	maxGetOrderTimeDuration = 24 * time.Hour
	minPollingDuration      = 1 * time.Minute
	defPollingDuration      = 5 * time.Minute
	maxPollingDuration      = 10 * time.Minute
)

var (
	channelMap map[int]map[chan<- *ServerMsg]int
	msgChan    chan *ServerMsg
)

type MsgOp struct {
	StoreID     int
	Chan2Listen chan<- *ServerMsg
	Chan2Close  chan int
}

type ServerMsg struct {
	Type    string
	StoreID int
	MsgData interface{}
}

func init() {
	utils.CallFuncAsync(routinueFunc)
}

func routinueFunc() {
	msgChan = make(chan *ServerMsg, 100)
	channelMap = make(map[int]map[chan<- *ServerMsg]int)

	for {
		msg, ok := <-msgChan
		if ok {
			switch msg.Type {
			case ServerMsgRegister:
				registerMsg := msg.MsgData.(*MsgOp)
				if channelMap[registerMsg.StoreID] == nil {
					channelMap[registerMsg.StoreID] = make(map[chan<- *ServerMsg]int)
				}
				channelMap[registerMsg.StoreID][registerMsg.Chan2Listen] = 1
			case ServerMsgUnregister:
				registerMsg := msg.MsgData.(*MsgOp)
				delete(channelMap[registerMsg.StoreID], registerMsg.Chan2Listen)
				close(registerMsg.Chan2Close)
			case ServerMsgNewOrder, ServerMsgFinishedPickup, ServerMsgKeyOrderStatusChanged, ServerMsgNewWait4ApproveAfsOrder, ServerMsgKeyAfsOrderStatusChanged:
				globals.SugarLogger.Debugf("msghub routinueFunc, msg:%s", utils.Format4Output(msg, true))
				utils.CallFuncAsync(func() {
					for chan2Send := range channelMap[msg.StoreID] {
						chan2Send <- msg
						delete(channelMap[msg.StoreID], chan2Send)
					}
				})
			}
		}
	}
}

func registerChan(storeID int, chan2Listen chan<- *ServerMsg) {
	msgChan <- &ServerMsg{
		Type: ServerMsgRegister,
		MsgData: &MsgOp{
			StoreID:     storeID,
			Chan2Listen: chan2Listen,
		},
	}
}

func unregisterChan(storeID int, chan2Listen chan<- *ServerMsg) {
	chan2Close := make(chan int)
	msgChan <- &ServerMsg{
		Type: ServerMsgUnregister,
		MsgData: &MsgOp{
			StoreID:     storeID,
			Chan2Listen: chan2Listen,
			Chan2Close:  chan2Close,
		},
	}
	<-chan2Close
}

func getPendingOrderList(storeID int, lastOrderTime time.Time, lastOrderSeqID int64) (vendorOrderIDs []string, err error) {
	if utils.IsTimeZero(lastOrderTime) || time.Now().Sub(lastOrderTime) > maxGetOrderTimeDuration {
		lastOrderTime = time.Now().Add(-maxGetOrderTimeDuration)
	}
	orderList, err := dao.GetStoreOrderAfterTime(dao.GetDB(), storeID, lastOrderTime, lastOrderSeqID)
	for _, v := range orderList {
		vendorOrderIDs = append(vendorOrderIDs, v.VendorOrderID)
	}
	return vendorOrderIDs, err
}

func GetMsg(ctx *jxcontext.Context, storeID int, lastOrderTime time.Time, lastOrderSeqID int64, msgTypeList []string, waitingSecond int) (msg *ServerMsg, err error) {
	vendorOrderIDs, err := getPendingOrderList(storeID, lastOrderTime, lastOrderSeqID)
	if err == nil {
		msg = &ServerMsg{
			Type:    ServerMsgNewOrder,
			StoreID: storeID,
			MsgData: 0,
		}
		if len(vendorOrderIDs) == 0 {
			chan2Listen := make(chan *ServerMsg, 1)
			registerChan(storeID, chan2Listen)
			pollingDuration := defPollingDuration
			if waitingSecond != 0 {
				pollingDuration = time.Duration(waitingSecond) * time.Second
				if globals.IsProductEnv() {
					if pollingDuration > maxPollingDuration {
						pollingDuration = maxPollingDuration
					} else if pollingDuration < minPollingDuration {
						pollingDuration = minPollingDuration
					}
				}
			}
			timer := time.NewTimer(pollingDuration)
			select {
			case msg2, ok := <-chan2Listen:
				timer.Stop()
				if ok {
					msg = msg2
				}
			case <-timer.C:
				unregisterChan(storeID, chan2Listen)
			}
			close(chan2Listen)
		} else {
			globals.SugarLogger.Debugf("GetMsg vendorOrderIDs:%s", utils.Format4Output(vendorOrderIDs, true))
			msg.MsgData = len(vendorOrderIDs)
		}
		storeMap, _ := permission.GetUserStoresResultMap(ctx.GetUserID())
		if storeMap[storeID] == 0 {
			msg.MsgData = 0
		}
	}
	return msg, err
}

func OnNewOrder(order *model.GoodsOrder) {
	globals.SugarLogger.Debugf("msghub OnNewOrder, order:%s", utils.Format4Output(order, true))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgNewOrder,
			StoreID: jxutils.GetSaleStoreIDFromOrder(order),
			MsgData: 1,
			// MsgData: []*model.GoodsOrderExt{
			// 	&model.GoodsOrderExt{
			// 		GoodsOrder: *order,
			// 	},
			// },
		}
	})
}

func OnFinishedPickup(order *model.GoodsOrder) {
	globals.SugarLogger.Debugf("msghub OnFinishedPickup, order:%s", utils.Format4Output(order, true))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgFinishedPickup,
			StoreID: jxutils.GetSaleStoreIDFromOrder(order),
			MsgData: 1,
			// MsgData: []*model.GoodsOrderExt{
			// 	&model.GoodsOrderExt{
			// 		GoodsOrder: *order,
			// 	},
			// },
		}
	})
}

func OnKeyOrderStatusChanged(order *model.GoodsOrder) {
	globals.SugarLogger.Debugf("msghub OnKeyOrderStatusChanged, order:%s", utils.Format4Output(order, true))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgKeyOrderStatusChanged,
			StoreID: jxutils.GetSaleStoreIDFromOrder(order),
			MsgData: order,
		}
	})
}

func OnNewWait4ApproveAfsOrder(order *model.AfsOrder) {
	globals.SugarLogger.Debugf("msghub OnNewWait4ApproveAfsOrder, order:%s", utils.Format4Output(order, true))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgNewWait4ApproveAfsOrder,
			StoreID: jxutils.GetSaleStoreIDFromAfsOrder(order),
			MsgData: order,
		}
	})
}

func OnKeyAfsOrderStatusChanged(order *model.AfsOrder) {
	globals.SugarLogger.Debugf("msghub OnKeyAfsOrderStatusChanged, order:%s", utils.Format4Output(order, true))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgKeyAfsOrderStatusChanged,
			StoreID: jxutils.GetSaleStoreIDFromAfsOrder(order),
			MsgData: order,
		}
	})
}