fusen/fusenapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
momo 2023-11-02 14:53:39 +08:00
commit 5efe255a77
11 changed files with 175 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
server/websocket/etc/websocket.yaml
View File

@ -25,4 +25,9 @@ BLMService:
#Url: "http://192.168.1.7:8999/LogoCombine"
Url: "http://18.119.109.254:8999/LogoCombine"
Unity:
Host: http://api.fusen.3718.cn:4050
Host: "http://api.fusen.3718.cn:4050"
Udp:
LocalAddr: "127.0.0.1"
LocalPort: 9100
RemoteAddr: "127.0.0.1"
RemotePort: 9101

6
server/websocket/internal/config/config.go
View File

@ -27,5 +27,11 @@ type Config struct {
}
Unity struct {
Host string
Udp struct {
LocalAddr string
LocalPort int
RemoteAddr string
RemotePort int
}
}
}

28
server/websocket/internal/logic/datatransferlogic.go
View File

@ -82,7 +82,7 @@ type wsConnectItem struct {
logic *DataTransferLogic //logic(基本属性，用于获取上下文，配置或者操作数据库)
closeChan chan struct{} //ws连接关闭chan(基本属性)
isClose bool //是否已经关闭(基本属性)
uniqueId string //ws连接唯一标识(基本属性)
wid string //ws连接唯一标识(基本属性)
inChan chan []byte //接受消息缓冲队列(基本属性)
outChan chan []byte //要发送回客户端的消息缓冲队列(基本属性)
mutex sync.Mutex //互斥锁(基本属性)
@ -170,7 +170,7 @@ func (l *DataTransferLogic) DataTransfer(req *types.DataTransferReq, w http.Resp
// 设置连接
func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo *auth.UserInfo, isFirefoxBrowser bool, userAgent, oldWid string) (wsConnectItem, error) {
//生成连接唯一标识（失败重试10次）
uniqueId, err := l.getUniqueId(userInfo, userAgent, 10)
wid, err := l.getUniqueId(userInfo, userAgent, 10)
if err != nil {
//发送获取唯一标识失败的消息
if isFirefoxBrowser {
@ -208,7 +208,7 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo *auth.Use
break
}
logx.Info("====复用旧的ws连接成功====")
uniqueId = oldWid
wid = oldWid
}
}
//默认过期时间
@ -224,7 +224,7 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo *auth.Use
logic: l,
closeChan: make(chan struct{}, 1),
isClose: false,
uniqueId: uniqueId,
wid: wid,
inChan: make(chan []byte, websocketInChanLen),
outChan: make(chan []byte, websocketOutChanLen),
mutex: sync.Mutex{},
@ -245,15 +245,15 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo *auth.Use
IsAllTemplateTag: 0,
}*/
//保存连接
mapConnPool.Store(uniqueId, ws)
mapConnPool.Store(wid, ws)
//累加统计连接数
increaseWebsocketConnectCount(userInfo.UserId, userInfo.GuestId)
//非白板用户，需要为这个用户建立map索引便于通过用户查询
createUserConnPoolElement(userInfo.UserId, userInfo.GuestId, uniqueId)
createUserConnPoolElement(userInfo.UserId, userInfo.GuestId, wid)
if isFirefoxBrowser {
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息：有待研究)
}
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, websocket_data.ConnectSuccessMsg{Wid: uniqueId}))
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, websocket_data.ConnectSuccessMsg{Wid: wid}))
return ws, nil
}
@ -304,7 +304,7 @@ func (w *wsConnectItem) heartbeat() {
case <-tick:
//看看token是否过期了
if w.connExpireTime > 0 && w.connExpireTime < time.Now().UTC().Unix() {
logx.Info("token过期，关闭连接:", w.uniqueId)
logx.Info("token过期，关闭连接:", w.wid)
w.close()
return
}
@ -313,7 +313,7 @@ func (w *wsConnectItem) heartbeat() {
w.debug = nil
}
if err := w.conn.WriteMessage(websocket.PongMessage, nil); err != nil {
logx.Error("发送心跳信息异常，关闭连接:", w.uniqueId, err)
logx.Error("发送心跳信息异常，关闭连接:", w.wid, err)
w.close()
return
}
@ -325,20 +325,22 @@ func (w *wsConnectItem) heartbeat() {
func (w *wsConnectItem) close() {
w.mutex.Lock()
defer w.mutex.Unlock()
logx.Info("###websocket:", w.uniqueId, " uid:", w.userId, " gid:", w.guestId, " is closing....")
logx.Info("###websocket:", w.wid, " uid:", w.userId, " gid:", w.guestId, " is closing....")
//发送关闭信息
_ = w.conn.WriteMessage(websocket.CloseMessage, nil)
w.conn.Close()
mapConnPool.Delete(w.uniqueId)
mapConnPool.Delete(w.wid)
if !w.isClose {
w.isClose = true
close(w.closeChan)
//删除用户级索引
deleteUserConnPoolElement(w.userId, w.guestId, w.uniqueId)
deleteUserConnPoolElement(w.userId, w.guestId, w.wid)
//减少连接数统计
decreaseWebsocketConnectCount(w.userId, w.guestId)
//通知unity取消任务
sendCancelRenderMsgToUnity(w.wid, time.Now().UTC().Unix())
}
logx.Info("###websocket:", w.uniqueId, " uid:", w.userId, " gid:", w.guestId, " is closed")
logx.Info("###websocket:", w.wid, " uid:", w.userId, " gid:", w.guestId, " is closed")
}
// 读取出口缓冲队列数据输出返回给浏览器端

24
server/websocket/internal/logic/rendernotifylogic.go
View File

@ -11,7 +11,6 @@ import (
"fusenapi/utils/basic"
"fusenapi/utils/file"
"fusenapi/utils/websocket_data"
"strings"
"time"
"github.com/zeromicro/go-zero/core/logx"
@ -39,19 +38,7 @@ func NewRenderNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Rend
// func (l *RenderNotifyLogic) AfterLogic(w http.ResponseWriter, r *http.Request, resp *basic.Response) {
// // httpx.OkJsonCtx(r.Context(), w, resp)
// }
func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) {
//统计unity处理数
decreaseUnityRequestCount(req.UserId, req.GuestId)
if req.Code != 0 {
//统计unity失败处理数
increaseUnityErrorCount(req.UserId, req.GuestId)
}
req.TaskId = strings.Trim(req.TaskId, " ")
if req.TaskId == "" {
logx.Error("渲染回调参数错误：任务标识")
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param task_id")
}
unityRenderEndTime := time.Now().UTC().UnixMilli()
//解析数据
var info websocket_data.ToUnityIdStruct
@ -59,6 +46,15 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *a
logx.Error("解析taskId错误")
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "failed to parse param taskId !!!!")
}
if info.RenderBeginTime > serverStartTime {
//logx.Info("任务时间：", info.RenderBeginTime, "服务器启动时间:", serverStartTime)
//统计unity处理数
decreaseUnityRequestCount(req.UserId, req.GuestId)
if req.Code != 0 {
//统计unity失败处理数
increaseUnityErrorCount(req.UserId, req.GuestId)
}
}
//重新赋值（很重要）
wid := info.Wid
requestId := info.RequestId
@ -99,7 +95,7 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *a
ws.sendRenderResultData(websocket_data.RenderImageRspMsg{
RequestId: requestId,
Image: uploadRes.ResourceUrl,
RenderProcessTime: websocket_data.RenderProcessTime{
RenderProcessTime: &websocket_data.RenderProcessTime{
UnityRenderTakesTime: fmt.Sprintf("%dms", unityRenderEndTime-unityRenderBeginTime),
UploadUnityRenderImageTakesTime: fmt.Sprintf("%dms", uploadUnityRenderImageTakesTime),
},

2
server/websocket/internal/logic/ws_ok_response.go
View File

@ -61,7 +61,7 @@ func (w *wsConnectItem) sendRenderDataToUnityStepResponseMessage(requestId strin
func (w *wsConnectItem) sendRenderResultData(data websocket_data.RenderImageRspMsg) {
//没开启debug
if w.debug == nil {
data.RenderProcessTime = websocket_data.RenderProcessTime{}
data.RenderProcessTime = nil
}
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, data))
}

96
server/websocket/internal/logic/ws_render_cancel_notify_unity.go Normal file
View File

@ -0,0 +1,96 @@
package logic
import (
"context"
"crypto/md5"
"encoding/hex"
"encoding/json"
"fmt"
"fusenapi/server/websocket/internal/config"
"github.com/zeromicro/go-zero/core/logx"
"net"
"time"
)
var (
//取消unity僵尸任务控制通道
cancelUnityCtlChan = make(chan cancelUnityCtlChanItem, 1000)
cancelRenderContextPanicMsg any = "cancel_render_context_panic_msg"
)
// 控制通道元素
type cancelUnityCtlChanItem struct {
Wid string `json:"wid"` //ws的唯一id
DeadlineTime int64 `json:"deadline_time"` //截断时间
Sign string `json:"sign"` //有效签名
}
// 取消渲染抛出的异常
func cancelRenderPanic() {
panic(cancelRenderContextPanicMsg)
}
// 判断是否是取消渲染的异常
func isCancelRenderPanic(err any) bool {
return err == cancelRenderContextPanicMsg
}
// 发送取消上下文消息给unity
func sendCancelRenderMsgToUnity(wid string, deadlineTime int64) {
data := cancelUnityCtlChanItem{
Wid: wid,
DeadlineTime: deadlineTime,
Sign: signMessage(wid, deadlineTime),
}
select {
case cancelUnityCtlChan <- data:
case <-time.After(time.Millisecond * 200):
logx.Error("sendCancelRenderMsgToUnity数据超时丢弃")
}
}
// 拨号udp
func DialUdp(ctx context.Context, config config.Config) error {
localAddr := &net.UDPAddr{IP: net.ParseIP(config.Unity.Udp.LocalAddr), Port: config.Unity.Udp.LocalPort}
remoteAddr := &net.UDPAddr{IP: net.ParseIP(config.Unity.Udp.RemoteAddr), Port: config.Unity.Udp.RemotePort}
conn, err := net.DialUDP("udp", localAddr, remoteAddr)
if err != nil {
return err
}
go ConsumeCancelUnityChanMessage(ctx, conn)
return nil
}
// 签名消息
func signMessage(wid string, deadlineTime int64) string {
h := md5.New()
h.Write([]byte(fmt.Sprintf("%s_fusen_control_unity_%d", wid, deadlineTime)))
return hex.EncodeToString(h.Sum(nil))
}
// 消费数据
func ConsumeCancelUnityChanMessage(ctx context.Context, conn *net.UDPConn) {
defer func() {
if err := recover(); err != nil {
logx.Error("ConsumeCancelUnityChanMessage 异常:", err)
}
}()
go func() {
select {
case <-ctx.Done():
panic(any("ConsumeCancelUnityChanMessage ctx deadline"))
}
}()
defer conn.Close()
for {
select {
case data := <-cancelUnityCtlChan:
d, _ := json.Marshal(data)
_, err := conn.Write(d)
if err != nil {
logx.Error("发送udp包通知Unity失败：", err)
continue
}
}
}
}

53
server/websocket/internal/logic/ws_render_image.go
View File

@ -50,9 +50,9 @@ func (r *renderProcessor) allocationMessage(w *wsConnectItem, data []byte) {
}
//颜色/模板标签/logo变更
ifCancelOldCtx := false
/*if renderImageData.RenderData.TemplateTag != w.extendRenderProperty.templateTag {
if renderImageData.RenderData.TemplateTag != w.extendRenderProperty.templateTag {
ifCancelOldCtx = true
}*/
}
if renderImageData.RenderData.TemplateTagColor.SelectedColorIndex != w.extendRenderProperty.selectColorIndex {
ifCancelOldCtx = true
}
@ -85,6 +85,11 @@ func (w *wsConnectItem) consumeRenderImageData() {
defer func() {
if err := recover(); err != nil {
logx.Error("func consumeRenderImageData panic:", err)
//如果是上下文取消渲染的异常
if isCancelRenderPanic(err) {
//通知unity取消任务
sendCancelRenderMsgToUnity(w.wid, time.Now().UTC().Unix())
}
}
}()
//限制并发
@ -95,20 +100,6 @@ func (w *wsConnectItem) consumeRenderImageData() {
case <-w.closeChan: //已关闭
return
case data := <-w.extendRenderProperty.renderChan: //消费数据
//标签不一样
/*if data.RenderData.TemplateTag != w.extendRenderProperty.templateTag {
//logx.Info("标签不一致，丢弃消息")
continue
}*/
//颜色不一致
if data.RenderData.TemplateTagColor.SelectedColorIndex != w.extendRenderProperty.selectColorIndex {
//logx.Info("颜色不一致，丢弃消息")
continue
}
//logo不一样
if data.RenderData.Logo != w.extendRenderProperty.Logo {
continue
}
limitChan <- struct{}{}
go func(d websocket_data.RenderImageReqMsg) {
defer func() {
@ -122,19 +113,23 @@ func (w *wsConnectItem) consumeRenderImageData() {
defer func() {
<-limitChan
}()
go func() {
defer func() {
if err := recover(); err != nil {
logx.Error("func renderImage panic:", err)
//如果不是无视上下文切换取消的(后面再开启)
/*if !d.IgnoreContextCancel {
go func() {
defer func() {
if err := recover(); err != nil {
logx.Error("func renderImage panic:", err)
}
}()
select {
case <-w.extendRenderProperty.renderCtx.Done():
//抛出取消渲染异常
cancelRenderPanic()
case <-tmpChan:
return
}
}()
select {
case <-w.extendRenderProperty.renderCtx.Done():
panic("检测到模板标签/颜色/logo变化，渲染取消旧的任务")
case <-tmpChan:
return
}
}()
}*/
w.renderImage(d)
}(data)
}
@ -269,7 +264,7 @@ func (w *wsConnectItem) renderImage(renderImageData websocket_data.RenderImageRe
w.sendRenderResultData(websocket_data.RenderImageRspMsg{
RequestId: renderImageData.RequestId,
Image: *resource.ResourceUrl,
RenderProcessTime: websocket_data.RenderProcessTime{
RenderProcessTime: &websocket_data.RenderProcessTime{
UnityRenderTakesTime: "cache",
UploadUnityRenderImageTakesTime: "cache",
},
@ -401,7 +396,7 @@ func (w *wsConnectItem) assembleRenderDataToUnity(taskId string, resolution int,
w.sendAssembleRenderDataStepResponseMessage(info.RequestId)
temId := websocket_data.ToUnityIdStruct{
TaskId: taskId,
Wid: w.uniqueId,
Wid: w.wid,
RequestId: info.RequestId,
RenderBeginTime: time.Now().UTC().UnixMilli(),
TemplateTag: info.RenderData.TemplateTag,

4
server/websocket/internal/logic/ws_statistics.go
View File

@ -34,6 +34,8 @@ type mapUserWsStatItem struct {
// 统计信息
var (
// 服务启动时间
serverStartTime = time.Now().UTC().UnixMilli()
//用户连接统计
mapUserWsStat = sync.Map{}
//消息控制通道的数据结构
@ -213,7 +215,7 @@ func ConsumeWebsocketStatData(ctx context.Context) {
go func() {
select {
case <-ctx.Done():
panic("ConsumeWebsocketStatData ctx deadline")
panic(any("ConsumeWebsocketStatData ctx deadline"))
}
}()
for {

2
server/websocket/internal/logic/ws_user_connect_pool.go
View File

@ -65,7 +65,7 @@ func ConsumeUserConnPoolCtlChanData(ctx context.Context) {
go func() {
select {
case <-ctx.Done():
panic("ConsumeUserConnPoolCtlChanData ctx deadline")
panic(any("ConsumeUserConnPoolCtlChanData ctx deadline"))
}
}()
var (

9
server/websocket/websocket.go
View File

@ -5,10 +5,10 @@ import (
"flag"
"fmt"
"fusenapi/server/websocket/internal/logic"
"net/http"
"fusenapi/utils/auth"
"fusenapi/utils/fsconfig"
"github.com/zeromicro/go-zero/core/logx"
"net/http"
"fusenapi/server/websocket/internal/config"
"fusenapi/server/websocket/internal/handler"
@ -39,6 +39,11 @@ func main() {
go logic.ConsumeUserConnPoolCtlChanData(ctx1)
//消费连接统计信息
go logic.ConsumeWebsocketStatData(ctx1)
//拨号udp消费控制unity取消僵尸任务的消息
if err := logic.DialUdp(ctx1, c); err != nil {
logx.Error("dail udp err:", err)
return
}
fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
server.Start()
}

11
utils/websocket_data/render_data.go
View File

@ -2,8 +2,9 @@ package websocket_data
// websocket接受要云渲染处理的数据
type RenderImageReqMsg struct {
RequestId string `json:"request_id"`
RenderData RenderData `json:"render_data"` //渲染主要参数
RequestId string `json:"request_id"`
IgnoreContextCancel bool `json:"ignore_context_cancel"` //是否无视切换上下文取消正在执行的渲染任务
RenderData RenderData `json:"render_data"` //渲染主要参数
}
type RenderData struct {
TemplateTag string `json:"template_tag"` //模板标签(必须)
@ -27,9 +28,9 @@ type TemplateTagColor struct {
// websocket发送渲染完的数据
type RenderImageRspMsg struct {
RequestId string `json:"request_id"`
Image string `json:"image"` //渲染结果图片
RenderProcessTime RenderProcessTime `json:"render_process_time"` //流程耗时
RequestId string `json:"request_id"`
Image string `json:"image"` //渲染结果图片
RenderProcessTime *RenderProcessTime `json:"render_process_time"` //流程耗时
}
type RenderProcessTime struct {
UnityRenderTakesTime string `json:"unity_render_takes_time"` //unity渲染用时