Merge branch 'develop' of https://gitee.com/fusenpack/fusenapi into develop

server/websocket/internal/logic/commonnotifylogic.go

@@ -5,7 +5,6 @@ import (
 	"fusenapi/constants"
 	"fusenapi/utils/auth"
 	"fusenapi/utils/basic"
-	"sync"
 	"time"
 
 	"fusenapi/server/websocket/internal/svc"
@@ -37,7 +36,7 @@ type commonConnectionNotFoundDataCacheChanItem struct {
 }
 
 // 放入缓冲队列
-func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
+func pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
 	select {
 	case commonConnectionNotFoundDataCacheChan <- data:
 		return
@@ -47,20 +46,26 @@ func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundD
 }
 
 // 取出元素
-func (l *CommonNotifyLogic) popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
+func popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
 	return <-commonConnectionNotFoundDataCacheChan
 }
 
-// 保证处理消息就一个循环在执行
-var consumeCommonCacheData sync.Once
-
 // 消费公共通知未处理的消息(目前是轮巡方式，待优化)
-func (l *CommonNotifyLogic) consumeCommonCacheData() {
+func ConsumeCommonCacheData(ctx context.Context) {
-	//单例
+	defer func() {
-	consumeCommonCacheData.Do(func() {
+		if err := recover(); err != nil {
+			logx.Error("consumeCommonCacheData panic :", err)
+		}
+	}()
+	go func() {
+		select {
+		case <-ctx.Done():
+			panic("consumeCommonCacheData ctx deadline")
+		}
+	}()
 	for {
 		time.Sleep(time.Millisecond * 200)
-			info := l.popCommonNotifyCache()
+		info := popCommonNotifyCache()
 		//查询websocket连接
 		value, ok := mapConnPool.Load(info.data.Wid)
 		//没有连接
@@ -68,7 +73,7 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 			info.retryTimes--
 			//大于0，则放回队列
 			if info.retryTimes > 0 {
-					l.pushCommonNotifyCache(info)
+				pushCommonNotifyCache(info)
 				continue
 			}
 			//否则直接丢弃消息
@@ -83,7 +88,6 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 		//发送
 		ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
 	}
-	})
 }
 
 // 处理进入前逻辑w,r
@@ -91,17 +95,20 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 // }
 
 func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) {
-	//websocket连接id不能为空
+	searchConnectType := 1
 	if req.Wid == "" {
-		return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "websocket connect id is empty")
+		if req.UserId == 0 && req.GuestId == 0 {
+			return resp.SetStatusWithMessage(basic.CodeOK, "用户信息或者连接标识必须保证至少有其中一个")
 		}
-	//触发消费公共未处理的消息（该方法是单例）
+		searchConnectType = 2
-	go l.consumeCommonCacheData()
+	}
+	switch searchConnectType {
+	case 1: //直接通过唯一标识发消息
 		//查询websocket连接
 		value, ok := mapConnPool.Load(req.Wid)
 		if !ok {
 			//没找到连接就放到公共缓冲队列
-		go l.pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
+			pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
 				retryTimes: 20, //重试20次
 				data: types.CommonNotifyReq{
 					Wid:  req.Wid,
@@ -117,6 +124,9 @@ func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *a
 			return resp.SetStatusWithMessage(basic.CodeServiceErr, "断言连接错误")
 		}
 		ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, req.Data))
+	case 2: //通过用户信息找连接发送
+		sendToOutChanByUserIndex(req.UserId, req.GuestId, (&wsConnectItem{}).respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, req.Data))
+	}
 	return resp.SetStatusWithMessage(basic.CodeOK, "success")
 }
 

server/websocket/internal/logic/datatransferlogic.go

@@ -148,8 +148,6 @@ func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request)
 		conn.Close()
 		return
 	}
-	//消费用户索引控制chan的数据
-	go consumeUserPoolData()
 	//循环读客户端信息
 	go ws.acceptBrowserMessage()
 	//消费出口数据并发送浏览器端
@@ -257,35 +255,41 @@ func sendToOutChanByUserIndex(userId, guestId int64, message []byte) {
 	}
 }
 
-// 消费用户索引池中的任务(单例)
+// 消费用户索引创建/删除/发送消息中的任务数据
-var consumeUserPoolDataOnce sync.Once
+func ConsumeUserPoolData(ctx context.Context) {
-
-func consumeUserPoolData() {
 	defer func() {
 		if err := recover(); err != nil {
 			logx.Error("consumeUserPoolData panic:", err)
 		}
 	}()
-	consumeUserPoolDataOnce.Do(func() {
+	go func() {
+		select {
+		case <-ctx.Done():
+			panic("ConsumeUserPoolData ctx deadline")
+		}
+	}()
 	for {
 		select {
 		case data := <-mapUserConnPoolCtlChan:
 			key := getmapUserConnPoolUniqueId(data.userId, data.guestId)
 			switch data.option {
 			case 2: //发送消息
-					logx.Info("通过用户id索引发送消息：", data.uniqueId)
+				logx.Info("通过用户id索引发送消息")
 				mapUserUniqueId, ok := mapUserConnPool[key]
 				if !ok {
+					logx.Info("通过用户id索引发送消息,连接不存在，用户索引key:", key)
 					continue
 				}
-					for _, uniqueId := range mapUserUniqueId {
+				for uniqueId, _ := range mapUserUniqueId {
 					//根据uniqueId查询原始池中连接
 					mapConnPoolVal, ok := mapConnPool.Load(uniqueId)
 					if !ok {
+						logx.Info("通过用户id索引发送消息,连接不存在，用户索引key:", key, "  原始uniqueId:", uniqueId)
 						continue
 					}
 					originConn, ok := mapConnPoolVal.(wsConnectItem)
 					if !ok {
+						logx.Error("通过用户id索引发送消息,断言原始连接失败，用户索引key:", key, "  原始uniqueId:", uniqueId)
 						continue
 					}
 					originConn.sendToOutChan(data.message)
@@ -308,7 +312,6 @@ func consumeUserPoolData() {
 			}
 		}
 	}
-	})
 }
 
 // 获取mapUserConnPool唯一id

server/websocket/internal/logic/ws_reuse_last_connect.go

@@ -61,8 +61,11 @@ func (r *reuseConnProcessor) allocationMessage(w *wsConnectItem, data []byte) {
 	}
 	//重新绑定
 	logx.Info("开始重新绑定websocket连接标识")
+	oldUniqueId := w.uniqueId
 	w.uniqueId = wid
 	mapConnPool.Store(wid, *w)
+	//删除用户id级别之前的索引
+	deleteUserConnPoolElement(w.userId, w.guestId, oldUniqueId)
 	//添加用户id级别索引
 	createUserConnPoolElement(w.userId, w.guestId, wid)
 	rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid)

server/websocket/internal/types/types.go

@@ -13,7 +13,9 @@ type RenderNotifyReq struct {
 }
 
 type CommonNotifyReq struct {
-	Wid  string                 `json:"wid"`  //websocket连接标识
+	Wid     string                 `json:"wid,optional"`      //websocket连接标识
+	UserId  int64                  `json:"user_id,optional"`  //用户id
+	GuestId int64                  `json:"guest_id,optional"` //游客id
 	Data    map[string]interface{} `json:"data"`              //后端与前端约定好的数据
 }
 

server/websocket/websocket.go

@@ -1,8 +1,10 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
+	"fusenapi/server/websocket/internal/logic"
 	"net/http"
 
 	"fusenapi/utils/auth"
@@ -28,6 +30,13 @@ func main() {
 
 	ctx := svc.NewServiceContext(c)
 	handler.RegisterHandlers(server, ctx)
+	ctx1 := context.Background()
+	ctx1, cancel := context.WithCancel(ctx1)
+	defer cancel()
+	//消费公共通知队列的数据
+	go logic.ConsumeCommonCacheData(ctx1)
+	//消费用户索引创建/删除/发送消息中的任务数据
+	go logic.ConsumeUserPoolData(ctx1)
 	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
 	server.Start()
 }

server_api/websocket.api

@@ -29,6 +29,8 @@ type RenderNotifyReq {
 }
 //通用回调接口
 type CommonNotifyReq {
-	Wid  string                 `json:"wid"`  //websocket连接标识
+	Wid     string                 `json:"wid,optional"`      //websocket连接标识
+	UserId  int64                  `json:"user_id,optional"`  //用户id
+	GuestId int64                  `json:"guest_id,optional"` //游客id
 	Data    map[string]interface{} `json:"data"`              //后端与前端约定好的数据
 }