fusenapi/server/websocket/internal/logic/datatransferlogic.go

405 lines
12 KiB
Go
Raw Normal View History

2023-07-24 08:07:05 +00:00
package logic
2023-08-23 07:08:45 +00:00
//websocket连接
2023-07-24 08:07:05 +00:00
import (
2023-07-26 08:00:19 +00:00
"bytes"
2023-08-23 08:13:14 +00:00
"encoding/hex"
2023-07-26 03:53:06 +00:00
"encoding/json"
2023-08-25 04:21:02 +00:00
"errors"
2023-07-26 03:53:06 +00:00
"fusenapi/constants"
2023-07-26 09:30:35 +00:00
"fusenapi/utils/auth"
2023-08-28 03:33:25 +00:00
"fusenapi/utils/basic"
2023-08-23 08:45:01 +00:00
"fusenapi/utils/encryption_decryption"
2023-08-24 06:24:04 +00:00
"fusenapi/utils/websocket_data"
2023-07-26 03:53:06 +00:00
"net/http"
2023-08-24 04:18:04 +00:00
"strings"
2023-07-26 03:53:06 +00:00
"sync"
"time"
2023-07-24 08:07:05 +00:00
"github.com/google/uuid"
2023-08-08 06:20:57 +00:00
"github.com/gorilla/websocket"
2023-07-24 08:07:05 +00:00
"context"
"fusenapi/server/websocket/internal/svc"
2023-08-08 06:20:57 +00:00
2023-07-24 08:07:05 +00:00
"github.com/zeromicro/go-zero/core/logx"
)
type DataTransferLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewDataTransferLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DataTransferLogic {
return &DataTransferLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
2023-07-26 03:53:06 +00:00
var (
2023-08-30 02:55:27 +00:00
//临时对象缓存池
2023-07-26 08:00:19 +00:00
buffPool = sync.Pool{
New: func() interface{} {
return bytes.Buffer{}
},
}
//升级websocket
2023-08-23 04:29:59 +00:00
upgrader = websocket.Upgrader{
2023-08-24 04:24:24 +00:00
//最大可读取大小 1M
ReadBufferSize: 1024,
//最大可写大小 1M
WriteBufferSize: 1024,
2023-07-26 08:00:19 +00:00
//握手超时时间15s
HandshakeTimeout: time.Second * 15,
2023-07-26 03:53:06 +00:00
//允许跨域
CheckOrigin: func(r *http.Request) bool {
return true
},
2023-08-28 07:50:18 +00:00
//写的缓冲队列
2023-07-27 11:30:40 +00:00
WriteBufferPool: &buffPool,
//是否支持压缩
2023-08-24 04:24:24 +00:00
EnableCompression: false,
2023-07-26 03:53:06 +00:00
}
2023-07-27 08:29:09 +00:00
//websocket连接存储
2023-07-26 03:53:06 +00:00
mapConnPool = sync.Map{}
2023-08-30 02:55:27 +00:00
//每个websocket连接入口缓冲队列长度
websocketInChanLen = 1000
//每个websocket连接出口缓冲队列长度
websocketOutChanLen = 1000
//渲染任务调度(添加任务/删除任务/修改任务属性缓冲队列长度该队列用于避免map并发读写冲突
renderImageTaskCtlChanLen = 500
//渲染任务缓冲队列长度
renderChanLen = 500
2023-07-26 03:53:06 +00:00
)
2023-07-24 08:07:05 +00:00
2023-07-26 09:06:53 +00:00
// 每个连接的连接基本属性
2023-07-26 03:53:06 +00:00
type wsConnectItem struct {
2023-08-24 04:04:58 +00:00
conn *websocket.Conn //websocket的连接(基本属性)
2023-08-30 03:25:20 +00:00
userAgent string //用户代理头信息(基本属性,用于重连标识验证因素之一)
2023-08-24 04:04:58 +00:00
logic *DataTransferLogic //logic(基本属性,用于获取上下文,配置或者操作数据库)
closeChan chan struct{} //ws连接关闭chan(基本属性)
isClose bool //是否已经关闭(基本属性)
uniqueId string //ws连接唯一标识(基本属性)
2023-08-28 07:50:18 +00:00
inChan chan []byte //接受消息缓冲队列(基本属性)
outChan chan []byte //要发送回客户端的消息缓冲队列(基本属性)
2023-08-24 04:04:58 +00:00
mutex sync.Mutex //互斥锁(基本属性)
userId int64 //用户id(基本属性)
guestId int64 //游客id(基本属性)
extendRenderProperty extendRenderProperty //扩展云渲染属性(扩展属性)
2023-07-26 03:53:06 +00:00
}
2023-07-24 08:07:05 +00:00
2023-08-18 08:21:45 +00:00
// 请求建立连接升级websocket协议
2023-08-16 06:16:33 +00:00
func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request) {
2023-08-18 08:20:45 +00:00
//把子协议携带的token设置到标准token头信息中
2023-08-18 09:52:19 +00:00
token := r.Header.Get("Sec-Websocket-Protocol")
2023-08-24 03:38:42 +00:00
//有token是正常用户无则是白板用户也可以连接
if token != "" {
r.Header.Set("Authorization", "Bearer "+token)
2023-08-24 03:43:16 +00:00
//设置Sec-Websocket-Protocol
upgrader.Subprotocols = []string{token}
2023-08-24 03:38:42 +00:00
}
2023-08-24 04:18:04 +00:00
//判断下是否火狐浏览器(获取浏览器第一条消息返回有收不到的bug需要延迟1秒)
userAgent := r.Header.Get("User-Agent")
//是否火狐浏览器
isFirefoxBrowser := false
if strings.Contains(userAgent, "Firefox") {
isFirefoxBrowser = true
}
2023-07-26 03:53:06 +00:00
//升级websocket
2023-08-23 04:29:59 +00:00
conn, err := upgrader.Upgrade(w, r, nil)
2023-07-26 03:53:06 +00:00
if err != nil {
logx.Error("http upgrade websocket err:", err)
return
}
2023-08-10 07:21:28 +00:00
//鉴权不成功后断开
2023-08-18 08:03:29 +00:00
var (
2023-08-07 11:13:16 +00:00
userInfo *auth.UserInfo
isAuth bool
)
2023-08-16 06:16:33 +00:00
isAuth, userInfo = l.checkAuth(r)
2023-07-27 07:46:07 +00:00
if !isAuth {
2023-08-18 09:33:42 +00:00
//未授权响应消息
2023-08-24 04:18:04 +00:00
l.unAuthResponse(conn, isFirefoxBrowser)
2023-08-23 10:34:01 +00:00
conn.Close()
2023-07-27 07:46:07 +00:00
return
2023-08-18 08:03:29 +00:00
}
2023-08-11 02:49:29 +00:00
//设置连接
2023-08-30 03:25:20 +00:00
ws, err := l.setConnPool(conn, userInfo, isFirefoxBrowser, userAgent)
2023-08-23 08:57:18 +00:00
if err != nil {
2023-08-23 10:34:01 +00:00
conn.Close()
2023-08-23 08:57:18 +00:00
return
}
2023-08-10 11:41:25 +00:00
//循环读客户端信息
2023-08-30 03:13:10 +00:00
go ws.acceptBrowserMessage()
//消费出口数据并发送浏览器端
go ws.consumeOutChanData()
//消费入口数据
go ws.consumeInChanData()
2023-08-10 11:41:25 +00:00
//操作连接中渲染任务的增加/删除
go ws.operationRenderTask()
2023-08-16 08:20:16 +00:00
//消费渲染缓冲队列
2023-08-30 03:13:10 +00:00
go ws.consumeRenderImageData()
2023-08-10 11:41:25 +00:00
//心跳
ws.heartbeat()
}
2023-08-11 02:49:29 +00:00
// 设置连接
2023-08-30 03:25:20 +00:00
func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo *auth.UserInfo, isFirefoxBrowser bool, userAgent string) (wsConnectItem, error) {
2023-08-25 04:21:02 +00:00
//生成连接唯一标识失败重试10次
2023-08-30 03:25:20 +00:00
uniqueId, err := l.getUniqueId(userInfo, userAgent, 10)
2023-08-23 08:57:18 +00:00
if err != nil {
//发送获取唯一标识失败的消息
l.sendGetUniqueIdErrResponse(conn)
return wsConnectItem{}, err
}
2023-07-26 03:53:06 +00:00
ws := wsConnectItem{
2023-07-26 09:06:53 +00:00
conn: conn,
2023-08-30 03:25:20 +00:00
userAgent: userAgent,
2023-08-16 06:09:03 +00:00
logic: l,
2023-07-26 09:06:53 +00:00
closeChan: make(chan struct{}, 1),
2023-08-30 03:25:20 +00:00
isClose: false,
uniqueId: uniqueId,
2023-08-30 02:55:27 +00:00
inChan: make(chan []byte, websocketInChanLen),
outChan: make(chan []byte, websocketOutChanLen),
2023-08-30 03:25:20 +00:00
mutex: sync.Mutex{},
2023-08-09 04:28:06 +00:00
userId: userInfo.UserId,
guestId: userInfo.GuestId,
2023-08-24 04:04:58 +00:00
extendRenderProperty: extendRenderProperty{
2023-08-21 08:19:30 +00:00
renderImageTask: make(map[string]*renderTask),
2023-08-30 02:55:27 +00:00
renderImageTaskCtlChan: make(chan renderImageControlChanItem, renderImageTaskCtlChanLen),
renderChan: make(chan []byte, renderChanLen),
2023-07-26 09:06:53 +00:00
},
2023-07-26 03:53:06 +00:00
}
//保存连接
2023-07-27 08:27:42 +00:00
mapConnPool.Store(uniqueId, ws)
2023-08-24 04:18:04 +00:00
if isFirefoxBrowser {
2023-08-18 08:23:04 +00:00
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
2023-08-24 04:18:04 +00:00
}
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId))
2023-08-23 08:57:18 +00:00
return ws, nil
2023-08-10 11:41:25 +00:00
}
2023-08-11 02:49:29 +00:00
// 获取唯一id
2023-08-30 03:25:20 +00:00
func (l *DataTransferLogic) getUniqueId(userInfo *auth.UserInfo, userAgent string, retryTimes int) (uniqueId string, err error) {
2023-08-25 04:21:02 +00:00
if retryTimes < 0 {
return "", errors.New("failed to get unique id")
}
2023-08-15 10:38:33 +00:00
//后面拼接上用户id
2023-08-30 03:25:20 +00:00
uniqueId = hex.EncodeToString([]byte(uuid.New().String())) + getUserJoinPart(userInfo.UserId, userInfo.GuestId, userAgent)
2023-08-25 04:21:02 +00:00
//存在则从新获取
2023-08-10 11:41:25 +00:00
if _, ok := mapConnPool.Load(uniqueId); ok {
2023-08-30 03:25:20 +00:00
uniqueId, err = l.getUniqueId(userInfo, userAgent, retryTimes-1)
2023-08-23 08:45:01 +00:00
if err != nil {
return "", err
}
}
//加密
uniqueId, err = encryption_decryption.CBCEncrypt(uniqueId)
if err != nil {
return "", err
2023-08-10 11:41:25 +00:00
}
2023-08-23 08:45:01 +00:00
return uniqueId, nil
2023-07-24 08:07:05 +00:00
}
2023-07-26 03:53:06 +00:00
2023-07-26 09:30:35 +00:00
// 鉴权
2023-08-16 06:16:33 +00:00
func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *auth.UserInfo) {
2023-07-26 09:30:35 +00:00
// 解析JWT token,并对空用户进行判断
2023-08-28 03:33:25 +00:00
userInfo, err := basic.ParseJwtToken(r, l.svcCtx)
2023-07-26 09:30:35 +00:00
if err != nil {
2023-07-27 04:32:03 +00:00
return false, nil
2023-07-26 09:30:35 +00:00
}
2023-08-24 03:27:53 +00:00
//白板用户
2023-08-28 03:33:25 +00:00
return true, userInfo
2023-07-26 09:30:35 +00:00
}
2023-08-18 09:33:42 +00:00
// 鉴权失败通知
2023-08-24 04:18:04 +00:00
func (l *DataTransferLogic) unAuthResponse(conn *websocket.Conn, isFirefoxBrowser bool) {
2023-08-18 09:33:42 +00:00
rsp := websocket_data.DataTransferData{
T: constants.WEBSOCKET_UNAUTH,
D: nil,
}
b, _ := json.Marshal(rsp)
2023-08-24 04:18:04 +00:00
if isFirefoxBrowser {
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
}
2023-08-18 09:33:42 +00:00
//先发一条正常信息
_ = conn.WriteMessage(websocket.TextMessage, b)
//发送关闭信息
_ = conn.WriteMessage(websocket.CloseMessage, nil)
2023-08-23 08:57:18 +00:00
}
// 获取唯一标识失败通知
func (l *DataTransferLogic) sendGetUniqueIdErrResponse(conn *websocket.Conn) {
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
rsp := websocket_data.DataTransferData{
T: constants.WEBSOCKET_GEN_UNIQUE_ID_ERR,
D: "err to gen unique id ",
}
b, _ := json.Marshal(rsp)
//先发一条正常信息
_ = conn.WriteMessage(websocket.TextMessage, b)
//发送关闭信息
_ = conn.WriteMessage(websocket.CloseMessage, nil)
2023-08-18 09:33:42 +00:00
}
2023-08-22 10:31:25 +00:00
// 心跳检测
2023-07-26 03:53:06 +00:00
func (w *wsConnectItem) heartbeat() {
2023-07-27 02:08:30 +00:00
tick := time.Tick(time.Second * 5)
2023-07-26 03:53:06 +00:00
for {
select {
case <-w.closeChan:
return
2023-07-26 10:00:46 +00:00
case <-tick:
2023-07-26 08:23:38 +00:00
//发送心跳信息
2023-07-26 09:39:56 +00:00
if err := w.conn.WriteMessage(websocket.PongMessage, nil); err != nil {
2023-07-27 08:27:42 +00:00
logx.Error("发送心跳信息异常,关闭连接:", w.uniqueId, err)
2023-08-24 03:56:10 +00:00
w.close()
2023-07-26 08:23:38 +00:00
return
}
2023-07-26 03:53:06 +00:00
}
}
}
2023-08-22 10:31:25 +00:00
// 关闭websocket连接
2023-07-26 03:53:06 +00:00
func (w *wsConnectItem) close() {
w.mutex.Lock()
defer w.mutex.Unlock()
2023-08-24 03:56:10 +00:00
logx.Info("###websocket:", w.uniqueId, " uid:", w.userId, " gid:", w.guestId, " is closing....")
2023-07-26 09:39:56 +00:00
//发送关闭信息
_ = w.conn.WriteMessage(websocket.CloseMessage, nil)
2023-07-26 03:53:06 +00:00
w.conn.Close()
2023-07-27 08:27:42 +00:00
mapConnPool.Delete(w.uniqueId)
2023-07-26 03:53:06 +00:00
if !w.isClose {
w.isClose = true
close(w.closeChan)
}
2023-08-24 03:56:10 +00:00
logx.Info("###websocket:", w.uniqueId, " uid:", w.userId, " gid:", w.guestId, " is closed")
2023-07-26 03:53:06 +00:00
}
2023-08-28 07:50:18 +00:00
// 读取出口缓冲队列数据输出返回给浏览器端
2023-08-30 03:13:10 +00:00
func (w *wsConnectItem) consumeOutChanData() {
2023-08-22 10:37:49 +00:00
defer func() {
if err := recover(); err != nil {
logx.Error("write loop panic:", err)
}
}()
2023-07-26 03:53:06 +00:00
for {
select {
case <-w.closeChan: //如果关闭了
return
case data := <-w.outChan:
2023-07-26 10:03:27 +00:00
if err := w.conn.WriteMessage(websocket.TextMessage, data); err != nil {
logx.Error("websocket write loop err:", err)
w.close()
return
}
2023-07-26 03:53:06 +00:00
}
}
}
2023-08-30 03:13:10 +00:00
// 消费websocket入口数据池中的数据
func (w *wsConnectItem) consumeInChanData() {
defer func() {
if err := recover(); err != nil {
logx.Error("send loop panic:", err)
}
}()
for {
select {
case <-w.closeChan:
return
case data := <-w.inChan:
w.dealwithReciveData(data)
}
}
}
// 接受浏览器端发来的消息并写入入口缓冲队列
func (w *wsConnectItem) acceptBrowserMessage() {
2023-08-22 10:37:49 +00:00
defer func() {
if err := recover(); err != nil {
logx.Error("read loop panic:", err)
}
}()
2023-07-26 03:53:06 +00:00
for {
select {
case <-w.closeChan: //如果关闭了
return
2023-08-28 07:34:30 +00:00
default: //收取消息
2023-07-26 09:55:14 +00:00
msgType, data, err := w.conn.ReadMessage()
2023-07-26 08:23:38 +00:00
if err != nil {
logx.Error("接受信息错误:", err)
//关闭连接
w.close()
return
}
2023-08-28 07:37:45 +00:00
//ping/pong/close的消息不处理
if msgType != websocket.PingMessage && msgType != websocket.PongMessage && msgType != websocket.CloseMessage {
2023-08-28 07:47:25 +00:00
w.sendToInChan(data)
2023-07-26 09:55:14 +00:00
}
2023-07-26 03:53:06 +00:00
}
}
}
2023-08-28 07:50:18 +00:00
// 把要传递给客户端的数据放入出口缓冲队列
2023-07-26 08:58:14 +00:00
func (w *wsConnectItem) sendToOutChan(data []byte) {
select {
case <-w.closeChan:
return
case w.outChan <- data:
2023-08-15 09:40:27 +00:00
return
2023-08-30 03:01:49 +00:00
case <-time.After(time.Millisecond * 200): //阻塞超过200ms丢弃
logx.Error("failed to send to out chan,time expired,data:", string(data))
2023-08-15 09:40:27 +00:00
return
2023-08-28 07:47:25 +00:00
}
}
// 发送接受到的消息到入口缓冲队列中
func (w *wsConnectItem) sendToInChan(data []byte) {
select {
case <-w.closeChan: //关闭了
return
case w.inChan <- data:
return
2023-08-30 02:55:27 +00:00
case <-time.After(time.Millisecond * 200): //200豪秒超时丢弃说明超过消费速度了
2023-08-30 06:02:29 +00:00
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_INCOME_CACHE_QUEUE_OVERFLOW, "send message is too frequent,the message is ignore by system:"+string(data)))
2023-08-28 07:47:25 +00:00
return
2023-07-26 04:27:34 +00:00
}
}
2023-08-22 10:31:25 +00:00
// 格式化为websocket标准返回格式
2023-08-23 07:08:45 +00:00
func (w *wsConnectItem) respondDataFormat(msgType constants.Websocket, data interface{}) []byte {
2023-08-07 03:25:06 +00:00
d := websocket_data.DataTransferData{
2023-07-27 09:33:50 +00:00
T: msgType,
D: data,
}
b, _ := json.Marshal(d)
return b
}
2023-08-28 07:50:18 +00:00
// 处理入口缓冲队列中不同类型的数据(分发处理)
2023-07-26 07:08:43 +00:00
func (w *wsConnectItem) dealwithReciveData(data []byte) {
2023-08-07 03:25:06 +00:00
var parseInfo websocket_data.DataTransferData
2023-07-26 07:08:43 +00:00
if err := json.Unmarshal(data, &parseInfo); err != nil {
2023-08-24 09:23:41 +00:00
w.incomeDataFormatErrResponse("invalid format of income message:" + string(data))
2023-08-15 10:14:44 +00:00
logx.Error("invalid format of websocket message:", err)
2023-07-26 07:08:43 +00:00
return
}
2023-07-26 08:47:53 +00:00
d, _ := json.Marshal(parseInfo.D)
2023-07-26 07:08:43 +00:00
//分消息类型给到不同逻辑处理,可扩展
switch parseInfo.T {
2023-08-07 05:17:23 +00:00
//图片渲染
case constants.WEBSOCKET_RENDER_IMAGE:
2023-08-16 08:20:16 +00:00
w.sendToRenderChan(d)
2023-08-11 02:49:29 +00:00
//刷新重连请求恢复上次连接的标识
2023-08-15 10:14:44 +00:00
case constants.WEBSOCKET_REQUEST_REUSE_LAST_CONNECT:
w.reuseLastConnect(d)
2023-07-26 07:08:43 +00:00
default:
2023-08-30 06:04:31 +00:00
logx.Error("未知消息类型uid:", w.userId, "gid:", w.guestId, "data:", string(data))
2023-07-26 07:08:43 +00:00
}
}