- first edition of long pulling

business/msghub/msghub.go

@@ -0,0 +1,153 @@
+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"
+)
+
+const (
+	maxGetOrderTimeDuration = 2 * time.Hour
+	pollingDuration         = 5 * 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:
+				globals.SugarLogger.Debugf("msghub routinueFunc, msg:%s", utils.Format4Output(msg, false))
+				go 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: ServerMsgRegister,
+		MsgData: &MsgOp{
+			StoreID:     storeID,
+			Chan2Listen: chan2Listen,
+			Chan2Close:  chan2Close,
+		},
+	}
+	<-chan2Close
+}
+
+func getPendingOrderList(storeID int, lastOrderTime time.Time, lastOrderSeqID int64) (orderList []*model.GoodsOrderExt, err error) {
+	if !jxutils.IsTimeEmpty(lastOrderTime) {
+		if 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) (msg *ServerMsg, err error) {
+	orderList, err := getPendingOrderList(storeID, lastOrderTime, lastOrderSeqID)
+	if err == nil {
+		if len(orderList) == 0 {
+			chan2Listen := make(chan *ServerMsg, 1)
+			registerChan(storeID, chan2Listen)
+			timer := time.NewTimer(pollingDuration)
+			select {
+			case msg2, ok := <-chan2Listen:
+				timer.Stop()
+				if ok {
+					msg = msg2
+				}
+			case <-timer.C:
+				unregisterChan(storeID, chan2Listen)
+			}
+			close(chan2Listen)
+		} else {
+			msg = &ServerMsg{
+				Type:    ServerMsgNewOrder,
+				StoreID: storeID,
+				MsgData: orderList,
+			}
+		}
+	}
+	return msg, err
+}
+
+func OnNewOrder(order *model.GoodsOrder) {
+	globals.SugarLogger.Debugf("msghub OnNewOrder, order:%s", utils.Format4Output(order, false))
+	go func() {
+		msgChan <- &ServerMsg{
+			Type:    ServerMsgNewOrder,
+			StoreID: jxutils.GetSaleStoreIDFromOrder(order),
+			MsgData: []*model.GoodsOrderExt{
+				&model.GoodsOrderExt{
+					GoodsOrder: *order,
+				},
+			},
+		}
+	}()
+}