jx-callback/business/msghub/msghub.go

package msghub

import (
	"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"
	ServerMsgKeyOrderStatusChanged = "keyOrderStatusChanged" // 重要订单状态变化
)

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, ServerMsgKeyOrderStatusChanged:
				globals.SugarLogger.Debugf("msghub routinueFunc, msg:%s", utils.Format4Output(msg, false))
				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) (orderList []*model.GoodsOrderExt, 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)
	return orderList, err
}

func GetMsg(ctx *jxcontext.Context, storeID int, lastOrderTime time.Time, lastOrderSeqID int64, msgTypeList []string, waitingSecond int) (msg *ServerMsg, err error) {
	orderList, err := getPendingOrderList(storeID, lastOrderTime, lastOrderSeqID)
	if err == nil {
		msg = &ServerMsg{
			Type:    ServerMsgNewOrder,
			StoreID: storeID,
			MsgData: 0,
		}
		if len(orderList) == 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.MsgData = msg2.MsgData
				}
			case <-timer.C:
				unregisterChan(storeID, chan2Listen)
			}
			close(chan2Listen)
		} else {
			msg.MsgData = len(orderList)
		}
	}
	return msg, err
}

func OnNewOrder(order *model.GoodsOrder) {
	globals.SugarLogger.Debugf("msghub OnNewOrder, order:%s", utils.Format4Output(order, false))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgNewOrder,
			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, false))
	utils.CallFuncAsync(func() {
		msgChan <- &ServerMsg{
			Type:    ServerMsgKeyOrderStatusChanged,
			StoreID: jxutils.GetSaleStoreIDFromOrder(order),
			MsgData: order,
			// MsgData: []*model.GoodsOrderExt{
			// 	&model.GoodsOrderExt{
			// 		GoodsOrder: *order,
			// 	},
			// },
		}
	})
}