This commit is contained in:
laodaming 2023-08-30 18:31:20 +08:00
parent 4ac4f7ee9d
commit 6a6f135b61
4 changed files with 31 additions and 26 deletions

View File

@ -6,23 +6,23 @@ import (
// 消息分发工厂 // 消息分发工厂
type allocationProcessorFactory interface { type allocationProcessorFactory interface {
allocationMessage(data []byte) allocationMessage(w *wsConnectItem, data []byte)
} }
var mapAllocationProcessor = make(map[constants.Websocket]allocationProcessorFactory) var mapAllocationProcessor = make(map[constants.Websocket]allocationProcessorFactory)
func (w *wsConnectItem) newAllocationProcessor(msgType constants.Websocket) allocationProcessorFactory { func (w *wsConnectItem) newAllocationProcessor(msgType constants.Websocket) allocationProcessorFactory {
if obj, ok := mapAllocationProcessor[msgType]; ok { if val, ok := mapAllocationProcessor[msgType]; ok {
return obj return val
} }
var obj allocationProcessorFactory var obj allocationProcessorFactory
switch msgType { switch msgType {
//图片渲染 //图片渲染
case constants.WEBSOCKET_RENDER_IMAGE: case constants.WEBSOCKET_RENDER_IMAGE:
obj = &renderProcessor{*w} obj = &renderProcessor{}
//刷新重连请求恢复上次连接的标识 //刷新重连请求恢复上次连接的标识
case constants.WEBSOCKET_REQUEST_REUSE_LAST_CONNECT: case constants.WEBSOCKET_REQUEST_REUSE_LAST_CONNECT:
obj = &reuseConnProcessor{*w} obj = &reuseConnProcessor{}
default: default:
} }

View File

@ -393,6 +393,10 @@ func (w *wsConnectItem) allocationProcessing(data []byte) {
d, _ := json.Marshal(parseInfo.D) d, _ := json.Marshal(parseInfo.D)
//获取工厂实例 //获取工厂实例
processor := w.newAllocationProcessor(parseInfo.T) processor := w.newAllocationProcessor(parseInfo.T)
if processor == nil {
logx.Error("未知消息类型:", string(data))
return
}
//执行工厂方法 //执行工厂方法
processor.allocationMessage(d) processor.allocationMessage(w, d)
} }

View File

@ -20,7 +20,6 @@ import (
// 渲染处理器 // 渲染处理器
type renderProcessor struct { type renderProcessor struct {
w wsConnectItem
} }
// 云渲染属性 // 云渲染属性
@ -49,12 +48,12 @@ type renderTask struct {
uploadUnityRenderImageTakesTime int64 //上传unity渲染结果图时间 uploadUnityRenderImageTakesTime int64 //上传unity渲染结果图时间
} }
func (r *renderProcessor) allocationMessage(data []byte) { func (r *renderProcessor) allocationMessage(w *wsConnectItem, data []byte) {
logx.Info("收到渲染任务消息:", string(data)) logx.Info("收到渲染任务消息:", string(data))
select { select {
case <-r.w.closeChan: //已经关闭 case <-w.closeChan: //已经关闭
return return
case r.w.extendRenderProperty.renderChan <- data: //发入到缓冲队列 case w.extendRenderProperty.renderChan <- data: //发入到缓冲队列
return return
case <-time.After(time.Second * 3): //三秒没进入缓冲队列就丢弃 case <-time.After(time.Second * 3): //三秒没进入缓冲队列就丢弃
return return

View File

@ -11,58 +11,60 @@ import (
// 复用连接处理器 // 复用连接处理器
type reuseConnProcessor struct { type reuseConnProcessor struct {
w wsConnectItem
} }
func (r *reuseConnProcessor) allocationMessage(data []byte) { func (r *reuseConnProcessor) allocationMessage(w *wsConnectItem, data []byte) {
logx.Info("收到请求恢复上次连接标识数据:", string(data)) logx.Info("收到请求恢复上次连接标识数据:", string(data))
var wid string var wid string
if err := json.Unmarshal(data, &wid); err != nil { if err := json.Unmarshal(data, &wid); err != nil {
logx.Error(" invalid format of wid :", wid) logx.Error(" invalid format of wid :", wid)
r.w.incomeDataFormatErrResponse("invalid format of wid") w.incomeDataFormatErrResponse("invalid format of wid")
return return
} }
//解密 //解密
decryptionWid, err := encryption_decryption.CBCDecrypt(wid) decryptionWid, err := encryption_decryption.CBCDecrypt(wid)
if err != nil { if err != nil {
r.w.reuseLastConnErrResponse("invalid wid") w.reuseLastConnErrResponse("invalid wid")
return return
} }
lendecryptionWid := len(decryptionWid) lendecryptionWid := len(decryptionWid)
//合成client后缀,不是同个后缀的不能复用 //合成client后缀,不是同个后缀的不能复用
userPart := getUserJoinPart(r.w.userId, r.w.guestId, r.w.userAgent) userPart := getUserJoinPart(w.userId, w.guestId, w.userAgent)
lenUserPart := len(userPart) lenUserPart := len(userPart)
if lendecryptionWid <= lenUserPart { if lendecryptionWid <= lenUserPart {
r.w.reuseLastConnErrResponse("length of client id is to short") w.reuseLastConnErrResponse("length of client id is to short")
return return
} }
//尾部不同不能复用 //尾部不同不能复用
if decryptionWid[lendecryptionWid-lenUserPart:] != userPart { if decryptionWid[lendecryptionWid-lenUserPart:] != userPart {
r.w.reuseLastConnErrResponse("the client id is not belong to you before") w.reuseLastConnErrResponse("the client id is not belong to you before")
return return
} }
//存在是不能给他申请重新绑定 //存在是不能给他申请重新绑定
if v, ok := mapConnPool.Load(wid); ok { if v, ok := mapConnPool.Load(wid); ok {
obj, ok := v.(wsConnectItem) obj, ok := v.(wsConnectItem)
if !ok { if !ok {
w.reuseLastConnErrResponse("连接断言失败")
logx.Error("连接断言失败") logx.Error("连接断言失败")
return
} }
//是当前自己占用(无需处理) //是当前自己占用(无需处理)
if obj.uniqueId == r.w.uniqueId { if obj.uniqueId == w.uniqueId {
rsp := r.w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid) rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid)
r.w.sendToOutChan(rsp) w.sendToOutChan(rsp)
return return
} else { } else {
r.w.reuseLastConnErrResponse("the wid is used by other people") w.reuseLastConnErrResponse("the wid is used by other people")
return return
} }
} }
//重新绑定 //重新绑定
r.w.uniqueId = wid logx.Info("开始重新绑定>>>>>")
mapConnPool.Store(wid, r.w) w.uniqueId = wid
rsp := r.w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid) mapConnPool.Store(wid, *w)
r.w.sendToOutChan(rsp) rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid)
return w.sendToOutChan(rsp)
logx.Info("重新绑定成功")
} }
// 获取用户拼接部分(复用标识用到) // 获取用户拼接部分(复用标识用到)