合并代码
This commit is contained in:
commit
e05fcd8248
18
constants/websocket.go
Normal file
18
constants/websocket.go
Normal file
|
@ -0,0 +1,18 @@
|
|||
package constants
|
||||
|
||||
type websocket string
|
||||
|
||||
// websocket消息类型
|
||||
const (
|
||||
//鉴权失败
|
||||
WEBSOCKET_UNAUTH = "unAuth"
|
||||
//ws连接成功
|
||||
WEBSOCKET_CONNECT_SUCCESS = "connect-success"
|
||||
//心跳信息
|
||||
WEBSOCKET_HEARTBEAT = "heartbeat"
|
||||
//图片渲染
|
||||
WEBSOCKET_RENDER_IMAGE = "render-image"
|
||||
)
|
||||
|
||||
// 云渲染通知需要的签名字符串
|
||||
const RENDER_NOTIFY_SIGN_KEY = "fusen-render-notify-%s-%d"
|
|
@ -1,6 +1,6 @@
|
|||
#! /bin/bash
|
||||
name=${1%%\\*}
|
||||
options=("backend" "product-template" "product-model")
|
||||
options=("backend" "product-template" "product-template-tag" "product-model")
|
||||
for option in "${options[@]}"; do
|
||||
if [ "$name" = "$option" ]; then
|
||||
echo $name
|
||||
|
|
|
@ -8,6 +8,7 @@ import (
|
|||
type FsProductTemplateTags struct {
|
||||
Id int64 `gorm:"primary_key;default:0;auto_increment;" json:"id"` // ID
|
||||
Title *string `gorm:"default:'';" json:"title"` // 标题
|
||||
CoverImg *string `gorm:"default:'';" json:"cover_img"` // 封面图
|
||||
Status *int64 `gorm:"default:0;" json:"status"` // 状态 1:可用
|
||||
CreateAt *int64 `gorm:"default:0;" json:"create_at"` // 创建时间
|
||||
}
|
||||
|
|
|
@ -23,3 +23,12 @@ func (pt *FsProductTemplateTagsModel) FindOne(ctx context.Context, id int64, fie
|
|||
err = db.Take(&resp).Error
|
||||
return resp, err
|
||||
}
|
||||
func (pt *FsProductTemplateTagsModel) GetList(ctx context.Context, page, limit int, orderBy string) (resp []FsProductTemplateTags, err error) {
|
||||
db := pt.db.WithContext(ctx).Model(&FsProductTemplateTags{}).Where("`status` = ?", 1)
|
||||
if orderBy != "" {
|
||||
db = db.Order(orderBy)
|
||||
}
|
||||
offset := (page - 1) * limit
|
||||
err = db.Offset(offset).Limit(limit).Find(&resp).Error
|
||||
return resp, err
|
||||
}
|
||||
|
|
|
@ -0,0 +1,8 @@
|
|||
Name: product-template-tag
|
||||
Host: 0.0.0.0
|
||||
Port: 9916
|
||||
SourceMysql: fusentest:XErSYmLELKMnf3Dh@tcp(110.41.19.98:3306)/fusentest
|
||||
Auth:
|
||||
AccessSecret: fusen2023
|
||||
AccessExpire: 2592000
|
||||
RefreshAfter: 1592000
|
12
server/product-template-tag/internal/config/config.go
Normal file
12
server/product-template-tag/internal/config/config.go
Normal file
|
@ -0,0 +1,12 @@
|
|||
package config
|
||||
|
||||
import (
|
||||
"fusenapi/server/product-template-tag/internal/types"
|
||||
"github.com/zeromicro/go-zero/rest"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
rest.RestConf
|
||||
SourceMysql string
|
||||
Auth types.Auth
|
||||
}
|
|
@ -0,0 +1,35 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"reflect"
|
||||
|
||||
"fusenapi/utils/basic"
|
||||
|
||||
"fusenapi/server/product-template-tag/internal/logic"
|
||||
"fusenapi/server/product-template-tag/internal/svc"
|
||||
"fusenapi/server/product-template-tag/internal/types"
|
||||
)
|
||||
|
||||
func GetProductTemplateTagsHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
var req types.GetProductTemplateTagsReq
|
||||
userinfo, err := basic.RequestParse(w, r, svcCtx, &req)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// 创建一个业务逻辑层实例
|
||||
l := logic.NewGetProductTemplateTagsLogic(r.Context(), svcCtx)
|
||||
|
||||
rl := reflect.ValueOf(l)
|
||||
basic.BeforeLogic(w, r, rl)
|
||||
|
||||
resp := l.GetProductTemplateTags(&req, userinfo)
|
||||
|
||||
if !basic.AfterLogic(w, r, rl, resp) {
|
||||
basic.NormalAfterLogic(w, r, resp)
|
||||
}
|
||||
}
|
||||
}
|
22
server/product-template-tag/internal/handler/routes.go
Normal file
22
server/product-template-tag/internal/handler/routes.go
Normal file
|
@ -0,0 +1,22 @@
|
|||
// Code generated by goctl. DO NOT EDIT.
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"fusenapi/server/product-template-tag/internal/svc"
|
||||
|
||||
"github.com/zeromicro/go-zero/rest"
|
||||
)
|
||||
|
||||
func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
|
||||
server.AddRoutes(
|
||||
[]rest.Route{
|
||||
{
|
||||
Method: http.MethodGet,
|
||||
Path: "/api/product-template/get_product_template_tags",
|
||||
Handler: GetProductTemplateTagsHandler(serverCtx),
|
||||
},
|
||||
},
|
||||
)
|
||||
}
|
|
@ -0,0 +1,55 @@
|
|||
package logic
|
||||
|
||||
import (
|
||||
"fusenapi/utils/auth"
|
||||
"fusenapi/utils/basic"
|
||||
|
||||
"context"
|
||||
|
||||
"fusenapi/server/product-template-tag/internal/svc"
|
||||
"fusenapi/server/product-template-tag/internal/types"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
)
|
||||
|
||||
type GetProductTemplateTagsLogic struct {
|
||||
logx.Logger
|
||||
ctx context.Context
|
||||
svcCtx *svc.ServiceContext
|
||||
}
|
||||
|
||||
func NewGetProductTemplateTagsLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetProductTemplateTagsLogic {
|
||||
return &GetProductTemplateTagsLogic{
|
||||
Logger: logx.WithContext(ctx),
|
||||
ctx: ctx,
|
||||
svcCtx: svcCtx,
|
||||
}
|
||||
}
|
||||
|
||||
// 处理进入前逻辑w,r
|
||||
// func (l *GetProductTemplateTagsLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) {
|
||||
// }
|
||||
|
||||
// 处理逻辑后 w,r 如:重定向, resp 必须重新处理
|
||||
// func (l *GetProductTemplateTagsLogic) AfterLogic(w http.ResponseWriter, r *http.Request, resp *basic.Response) {
|
||||
// // httpx.OkJsonCtx(r.Context(), w, resp)
|
||||
// }
|
||||
|
||||
func (l *GetProductTemplateTagsLogic) GetProductTemplateTags(req *types.GetProductTemplateTagsReq, userinfo *auth.UserInfo) (resp *basic.Response) {
|
||||
if req.Limit <= 0 || req.Limit > 100 {
|
||||
req.Limit = 4
|
||||
}
|
||||
productTemplateTags, err := l.svcCtx.AllModels.FsProductTemplateTags.GetList(l.ctx, 1, req.Limit, "`id` DESC")
|
||||
if err != nil {
|
||||
logx.Error(err)
|
||||
return resp.SetStatusWithMessage(basic.CodeDbSqlErr, "failed to get template tags")
|
||||
}
|
||||
list := make([]types.GetProductTemplateTagsRsp, 0, len(productTemplateTags))
|
||||
for _, v := range productTemplateTags {
|
||||
list = append(list, types.GetProductTemplateTagsRsp{
|
||||
Tag: *v.Title,
|
||||
Cover: *v.CoverImg,
|
||||
})
|
||||
}
|
||||
return resp.SetStatusWithMessage(basic.CodeOK, "success", list)
|
||||
}
|
61
server/product-template-tag/internal/svc/servicecontext.go
Normal file
61
server/product-template-tag/internal/svc/servicecontext.go
Normal file
|
@ -0,0 +1,61 @@
|
|||
package svc
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"fusenapi/server/product-template-tag/internal/config"
|
||||
"net/http"
|
||||
|
||||
"fusenapi/initalize"
|
||||
"fusenapi/model/gmodel"
|
||||
|
||||
"github.com/golang-jwt/jwt"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type ServiceContext struct {
|
||||
Config config.Config
|
||||
|
||||
MysqlConn *gorm.DB
|
||||
AllModels *gmodel.AllModelsGen
|
||||
}
|
||||
|
||||
func NewServiceContext(c config.Config) *ServiceContext {
|
||||
|
||||
return &ServiceContext{
|
||||
Config: c,
|
||||
MysqlConn: initalize.InitMysql(c.SourceMysql),
|
||||
AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)),
|
||||
}
|
||||
}
|
||||
|
||||
func (svcCtx *ServiceContext) ParseJwtToken(r *http.Request) (jwt.MapClaims, error) {
|
||||
AuthKey := r.Header.Get("Authorization")
|
||||
if AuthKey == "" {
|
||||
return nil, nil
|
||||
}
|
||||
AuthKey = AuthKey[7:]
|
||||
|
||||
if len(AuthKey) <= 50 {
|
||||
return nil, errors.New(fmt.Sprint("Error parsing token, len:", len(AuthKey)))
|
||||
}
|
||||
|
||||
token, err := jwt.Parse(AuthKey, func(token *jwt.Token) (interface{}, error) {
|
||||
// 检查签名方法是否为 HS256
|
||||
if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
|
||||
return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
|
||||
}
|
||||
// 返回用于验证签名的密钥
|
||||
return []byte(svcCtx.Config.Auth.AccessSecret), nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.New(fmt.Sprint("Error parsing token:", err))
|
||||
}
|
||||
|
||||
// 验证成功返回
|
||||
if claims, ok := token.Claims.(jwt.MapClaims); ok && token.Valid {
|
||||
return claims, nil
|
||||
}
|
||||
|
||||
return nil, errors.New(fmt.Sprint("Invalid token", err))
|
||||
}
|
84
server/product-template-tag/internal/types/types.go
Normal file
84
server/product-template-tag/internal/types/types.go
Normal file
|
@ -0,0 +1,84 @@
|
|||
// Code generated by goctl. DO NOT EDIT.
|
||||
package types
|
||||
|
||||
import (
|
||||
"fusenapi/utils/basic"
|
||||
)
|
||||
|
||||
type GetProductTemplateTagsReq struct {
|
||||
Limit int `form:"limit"`
|
||||
}
|
||||
|
||||
type GetProductTemplateTagsRsp struct {
|
||||
Tag string `json:"tag"`
|
||||
Cover string `json:"cover"`
|
||||
}
|
||||
|
||||
type Request struct {
|
||||
}
|
||||
|
||||
type Response struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"msg"`
|
||||
Data interface{} `json:"data"`
|
||||
}
|
||||
|
||||
type Auth struct {
|
||||
AccessSecret string `json:"accessSecret"`
|
||||
AccessExpire int64 `json:"accessExpire"`
|
||||
RefreshAfter int64 `json:"refreshAfter"`
|
||||
}
|
||||
|
||||
type File struct {
|
||||
Filename string `fsfile:"filename"`
|
||||
Header map[string][]string `fsfile:"header"`
|
||||
Size int64 `fsfile:"size"`
|
||||
Data []byte `fsfile:"data"`
|
||||
}
|
||||
|
||||
type Meta struct {
|
||||
TotalCount int64 `json:"totalCount"`
|
||||
PageCount int64 `json:"pageCount"`
|
||||
CurrentPage int `json:"currentPage"`
|
||||
PerPage int `json:"perPage"`
|
||||
}
|
||||
|
||||
// Set 设置Response的Code和Message值
|
||||
func (resp *Response) Set(Code int, Message string) *Response {
|
||||
return &Response{
|
||||
Code: Code,
|
||||
Message: Message,
|
||||
}
|
||||
}
|
||||
|
||||
// Set 设置整个Response
|
||||
func (resp *Response) SetWithData(Code int, Message string, Data interface{}) *Response {
|
||||
return &Response{
|
||||
Code: Code,
|
||||
Message: Message,
|
||||
Data: Data,
|
||||
}
|
||||
}
|
||||
|
||||
// SetStatus 设置默认StatusResponse(内部自定义) 默认msg, 可以带data, data只使用一个参数
|
||||
func (resp *Response) SetStatus(sr *basic.StatusResponse, data ...interface{}) *Response {
|
||||
newResp := &Response{
|
||||
Code: sr.Code,
|
||||
}
|
||||
if len(data) == 1 {
|
||||
newResp.Data = data[0]
|
||||
}
|
||||
return newResp
|
||||
}
|
||||
|
||||
// SetStatusWithMessage 设置默认StatusResponse(内部自定义) 非默认msg, 可以带data, data只使用一个参数
|
||||
func (resp *Response) SetStatusWithMessage(sr *basic.StatusResponse, msg string, data ...interface{}) *Response {
|
||||
newResp := &Response{
|
||||
Code: sr.Code,
|
||||
Message: msg,
|
||||
}
|
||||
if len(data) == 1 {
|
||||
newResp.Data = data[0]
|
||||
}
|
||||
return newResp
|
||||
}
|
36
server/product-template-tag/product-template-tag.go
Normal file
36
server/product-template-tag/product-template-tag.go
Normal file
|
@ -0,0 +1,36 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"fusenapi/utils/auth"
|
||||
|
||||
"fusenapi/server/product-template-tag/internal/config"
|
||||
"fusenapi/server/product-template-tag/internal/handler"
|
||||
"fusenapi/server/product-template-tag/internal/svc"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/conf"
|
||||
"github.com/zeromicro/go-zero/rest"
|
||||
)
|
||||
|
||||
var configFile = flag.String("f", "etc/product-template-tag.yaml", "the config file")
|
||||
|
||||
func main() {
|
||||
flag.Parse()
|
||||
|
||||
var c config.Config
|
||||
conf.MustLoad(*configFile, &c)
|
||||
c.Timeout = int64(time.Second * 15)
|
||||
server := rest.MustNewServer(c.RestConf, rest.WithCustomCors(auth.FsCors, func(w http.ResponseWriter) {
|
||||
}))
|
||||
defer server.Stop()
|
||||
|
||||
ctx := svc.NewServiceContext(c)
|
||||
handler.RegisterHandlers(server, ctx)
|
||||
|
||||
fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
|
||||
server.Start()
|
||||
}
|
|
@ -1,6 +1,6 @@
|
|||
Name: websocket
|
||||
Host: 0.0.0.0
|
||||
Port: 8888
|
||||
Port: 9914
|
||||
SourceMysql: fusentest:XErSYmLELKMnf3Dh@tcp(110.41.19.98:3306)/fusentest
|
||||
Auth:
|
||||
AccessSecret: fusen2023
|
||||
|
|
|
@ -1,10 +1,19 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"fusenapi/constants"
|
||||
"fusenapi/server/websocket/internal/logic"
|
||||
"fusenapi/server/websocket/internal/svc"
|
||||
"fusenapi/server/websocket/internal/types"
|
||||
"fusenapi/utils/auth"
|
||||
"github.com/google/uuid"
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
|
@ -16,43 +25,246 @@ var (
|
|||
},
|
||||
}
|
||||
//连接map池
|
||||
mapConn = sync.Map{}
|
||||
mapConnPool = sync.Map{}
|
||||
)
|
||||
|
||||
// 每个连接的连接属性
|
||||
type wsConnectItem struct {
|
||||
conn *websocket.Conn //websocket的连接
|
||||
renImage sync.Map //需要渲染的图片
|
||||
conn *websocket.Conn //websocket的连接
|
||||
closeChan chan struct{} //关闭chan
|
||||
isClose bool //是否已经关闭
|
||||
flag string
|
||||
inChan chan []byte //接受消息缓冲通道
|
||||
outChan chan []byte //发送回客户端的消息
|
||||
mutex sync.Mutex
|
||||
renderImage map[string]struct{} //需要渲染的图片
|
||||
}
|
||||
|
||||
func DataTransferHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
/*// 解析JWT token,并对空用户进行判断
|
||||
// 创建一个业务逻辑层实例
|
||||
var req types.DataTransferReq
|
||||
l := logic.NewDataTransferLogic(r.Context(), svcCtx)
|
||||
l.DataTransfer(&req, nil)
|
||||
//升级websocket
|
||||
conn, err := upgrade.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
logx.Error("http upgrade websocket err:", err)
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
rsp := types.DataTransferRsp{}
|
||||
isAuth := true
|
||||
// 解析JWT token,并对空用户进行判断
|
||||
claims, err := svcCtx.ParseJwtToken(r)
|
||||
// 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息
|
||||
if err != nil {
|
||||
logx.Info("unauthorized:", err.Error()) // 记录错误日志
|
||||
w.Write([]byte("connect failed:unauthorized"))
|
||||
return
|
||||
rsp.T = constants.WEBSOCKET_UNAUTH
|
||||
rsp.D = "unAuth"
|
||||
b, _ := json.Marshal(rsp)
|
||||
_ = conn.WriteMessage(websocket.TextMessage, b)
|
||||
isAuth = false
|
||||
}
|
||||
var userInfo *auth.UserInfo
|
||||
if claims != nil {
|
||||
// 从token中获取对应的用户信息
|
||||
userInfo, err = auth.GetUserInfoFormMapClaims(claims)
|
||||
_, err = auth.GetUserInfoFormMapClaims(claims)
|
||||
// 如果获取用户信息出错,则返回未授权的JSON响应并记录错误消息
|
||||
if err != nil {
|
||||
return
|
||||
rsp.T = constants.WEBSOCKET_UNAUTH
|
||||
rsp.D = "unAuth!!"
|
||||
b, _ := json.Marshal(rsp)
|
||||
_ = conn.WriteMessage(websocket.TextMessage, b)
|
||||
isAuth = false
|
||||
}
|
||||
} else {
|
||||
// 如果claims为nil,则认为用户身份为白板用户
|
||||
w.Write([]byte("connect failed:unauthorized!!"))
|
||||
return
|
||||
rsp.T = constants.WEBSOCKET_UNAUTH
|
||||
rsp.D = "unAuth!!!"
|
||||
b, _ := json.Marshal(rsp)
|
||||
_ = conn.WriteMessage(websocket.TextMessage, b)
|
||||
isAuth = false
|
||||
}
|
||||
//升级websocket
|
||||
conn, err := upgrade.Upgrade(w, r, r.Header)
|
||||
if err != nil {
|
||||
logx.Error("http upgrade websocket err:", err)
|
||||
w.Write([]byte("http upgrade websocket err"))
|
||||
return
|
||||
}*/
|
||||
//不是授权的连接(10秒后关闭)
|
||||
if !isAuth {
|
||||
select {
|
||||
case <-time.After(time.Second * 10):
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
//生成连接唯一标识
|
||||
flag := uuid.New().String() + time.Now().Format("20060102150405")
|
||||
ws := wsConnectItem{
|
||||
conn: conn,
|
||||
flag: flag,
|
||||
closeChan: make(chan struct{}, 1),
|
||||
inChan: make(chan []byte, 100),
|
||||
outChan: make(chan []byte, 100),
|
||||
renderImage: make(map[string]struct{}),
|
||||
isClose: false,
|
||||
}
|
||||
//保存连接
|
||||
mapConnPool.Store(flag, ws)
|
||||
defer ws.close()
|
||||
//把连接成功消息发回去
|
||||
rsp.T = constants.WEBSOCKET_CONNECT_SUCCESS
|
||||
rsp.D = flag
|
||||
b, _ := json.Marshal(rsp)
|
||||
conn.WriteMessage(websocket.TextMessage, b)
|
||||
//循环读客户端信息
|
||||
go ws.readLoop()
|
||||
//循环把数据发送给客户端
|
||||
go ws.writeLoop()
|
||||
//推消息到云渲染
|
||||
go ws.sendLoop()
|
||||
//心跳
|
||||
ws.heartbeat()
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
// 心跳
|
||||
func (w *wsConnectItem) heartbeat() {
|
||||
rsp := types.DataTransferRsp{
|
||||
T: constants.WEBSOCKET_HEARTBEAT,
|
||||
D: "",
|
||||
}
|
||||
b, _ := json.Marshal(rsp)
|
||||
for {
|
||||
time.Sleep(time.Second * 10)
|
||||
select {
|
||||
case <-w.closeChan:
|
||||
return
|
||||
default:
|
||||
|
||||
}
|
||||
//发送心跳信息
|
||||
if err := w.conn.WriteMessage(websocket.TextMessage, b); err != nil {
|
||||
logx.Error("发送心跳信息异常,关闭连接:", w.flag, err)
|
||||
w.close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 关闭连接
|
||||
func (w *wsConnectItem) close() {
|
||||
w.mutex.Lock()
|
||||
defer w.mutex.Unlock()
|
||||
logx.Info("websocket:", w.flag, " is closing...")
|
||||
w.conn.Close()
|
||||
mapConnPool.Delete(w.flag)
|
||||
if !w.isClose {
|
||||
w.isClose = true
|
||||
close(w.closeChan)
|
||||
close(w.outChan)
|
||||
close(w.inChan)
|
||||
}
|
||||
logx.Info("websocket:", w.flag, " is closed")
|
||||
}
|
||||
|
||||
func (w *wsConnectItem) writeLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-w.closeChan: //如果关闭了
|
||||
return
|
||||
case data := <-w.outChan:
|
||||
w.conn.WriteMessage(websocket.TextMessage, data)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 接受客户端发来的消息
|
||||
func (w *wsConnectItem) readLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-w.closeChan: //如果关闭了
|
||||
return
|
||||
default:
|
||||
|
||||
}
|
||||
_, data, err := w.conn.ReadMessage()
|
||||
if err != nil {
|
||||
logx.Error("接受信息错误:", err)
|
||||
//关闭连接
|
||||
w.close()
|
||||
return
|
||||
}
|
||||
//消息传入缓冲通道
|
||||
w.inChan <- data
|
||||
}
|
||||
}
|
||||
|
||||
// 把收到的消息发往不同的地方处理
|
||||
func (w *wsConnectItem) sendLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-w.closeChan:
|
||||
return
|
||||
case data := <-w.inChan:
|
||||
w.dealwithReciveData(data)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 处理接受到的数据
|
||||
func (w *wsConnectItem) dealwithReciveData(data []byte) {
|
||||
var parseInfo types.DataTransferReq
|
||||
if err := json.Unmarshal(data, &parseInfo); err != nil {
|
||||
logx.Error("invalid format of websocket message")
|
||||
return
|
||||
}
|
||||
switch parseInfo.T {
|
||||
//图片渲染
|
||||
case constants.WEBSOCKET_RENDER_IMAGE:
|
||||
var renderImageData []types.RenderImageReqMsg
|
||||
if err := json.Unmarshal([]byte(parseInfo.D), &renderImageData); err != nil {
|
||||
logx.Error("invalid format of websocket render image message", err)
|
||||
return
|
||||
}
|
||||
logx.Info("收到请求云渲染图片数据:", renderImageData)
|
||||
w.mutex.Lock()
|
||||
defer w.mutex.Unlock()
|
||||
//把需要渲染的图片加进去
|
||||
for _, v := range renderImageData {
|
||||
key := w.getRenderImageMapKey(v.ProductId, v.SizeId, v.TemplateId)
|
||||
w.renderImage[key] = struct{}{}
|
||||
}
|
||||
default:
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
// 把渲染好的数据放入outchan
|
||||
func (w *wsConnectItem) setOutRenderImage(req types.RenderNotifyReq) {
|
||||
w.mutex.Lock()
|
||||
defer w.mutex.Unlock()
|
||||
for _, notifyItem := range req.NotifyList {
|
||||
renderKey := w.getRenderImageMapKey(notifyItem.ProductId, notifyItem.SizeId, notifyItem.TemplateId)
|
||||
//查询
|
||||
_, ok := w.renderImage[renderKey]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
responseData := types.RenderImageRspMsg{
|
||||
ProductId: notifyItem.ProductId,
|
||||
SizeId: notifyItem.SizeId,
|
||||
TemplateId: notifyItem.TemplateId,
|
||||
Source: "我是渲染资源",
|
||||
}
|
||||
b, _ := json.Marshal(responseData)
|
||||
select {
|
||||
case <-w.closeChan:
|
||||
return
|
||||
case w.outChan <- b:
|
||||
logx.Info("notify send render result to out chan")
|
||||
}
|
||||
//删掉已经处理的渲染任务
|
||||
delete(w.renderImage, renderKey)
|
||||
}
|
||||
}
|
||||
|
||||
// 获取需要渲染图片的map key
|
||||
func (w *wsConnectItem) getRenderImageMapKey(productId, sizeId, templateId int64) string {
|
||||
return fmt.Sprintf("%d-%d-%d", productId, sizeId, templateId)
|
||||
}
|
||||
|
|
75
server/websocket/internal/handler/rendernotifyhandler.go
Normal file
75
server/websocket/internal/handler/rendernotifyhandler.go
Normal file
|
@ -0,0 +1,75 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"fusenapi/constants"
|
||||
"fusenapi/utils/basic"
|
||||
"github.com/zeromicro/go-zero/rest/httpx"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"fusenapi/server/websocket/internal/svc"
|
||||
"fusenapi/server/websocket/internal/types"
|
||||
)
|
||||
|
||||
func RenderNotifyHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
var req types.RenderNotifyReq
|
||||
_, err := basic.RequestParse(w, r, svcCtx, &req)
|
||||
if err != nil {
|
||||
httpx.OkJsonCtx(r.Context(), w, basic.Response{
|
||||
Code: basic.CodeRequestParamsErr.Code,
|
||||
Message: "err param",
|
||||
Data: nil,
|
||||
})
|
||||
return
|
||||
}
|
||||
if len(req.NotifyList) == 0 {
|
||||
httpx.OkJsonCtx(r.Context(), w, basic.Response{
|
||||
Code: basic.CodeRequestParamsErr.Code,
|
||||
Message: "invalid param,notify list is empty",
|
||||
Data: nil,
|
||||
})
|
||||
return
|
||||
}
|
||||
if time.Now().Unix()-120 > req.Time /*|| req.Time > time.Now().Unix() */ {
|
||||
httpx.OkJsonCtx(r.Context(), w, basic.Response{
|
||||
Code: basic.CodeRequestParamsErr.Code,
|
||||
Message: "invalid param,time is expired",
|
||||
Data: nil,
|
||||
})
|
||||
return
|
||||
}
|
||||
//验证签名 sha256
|
||||
notifyByte, _ := json.Marshal(req.NotifyList)
|
||||
h := sha256.New()
|
||||
h.Write([]byte(fmt.Sprintf(constants.RENDER_NOTIFY_SIGN_KEY, string(notifyByte), req.Time)))
|
||||
signHex := h.Sum(nil)
|
||||
sign := hex.EncodeToString(signHex)
|
||||
if req.Sign != sign {
|
||||
httpx.OkJsonCtx(r.Context(), w, basic.Response{
|
||||
Code: basic.CodeRequestParamsErr.Code,
|
||||
Message: "invalid sign",
|
||||
Data: nil,
|
||||
})
|
||||
return
|
||||
}
|
||||
//遍历链接
|
||||
mapConnPool.Range(func(key, value any) bool {
|
||||
ws, ok := value.(wsConnectItem)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
ws.setOutRenderImage(req)
|
||||
return true
|
||||
})
|
||||
httpx.OkJsonCtx(r.Context(), w, basic.Response{
|
||||
Code: 200,
|
||||
Message: "success",
|
||||
Data: nil,
|
||||
})
|
||||
}
|
||||
}
|
|
@ -17,6 +17,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
|
|||
Path: "/api/websocket/data_transfer",
|
||||
Handler: DataTransferHandler(serverCtx),
|
||||
},
|
||||
{
|
||||
Method: http.MethodPost,
|
||||
Path: "/api/websocket/render_notify",
|
||||
Handler: RenderNotifyHandler(serverCtx),
|
||||
},
|
||||
},
|
||||
)
|
||||
}
|
||||
|
|
43
server/websocket/internal/logic/rendernotifylogic.go
Normal file
43
server/websocket/internal/logic/rendernotifylogic.go
Normal file
|
@ -0,0 +1,43 @@
|
|||
package logic
|
||||
|
||||
import (
|
||||
"fusenapi/utils/auth"
|
||||
"fusenapi/utils/basic"
|
||||
|
||||
"context"
|
||||
|
||||
"fusenapi/server/websocket/internal/svc"
|
||||
"fusenapi/server/websocket/internal/types"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
)
|
||||
|
||||
type RenderNotifyLogic struct {
|
||||
logx.Logger
|
||||
ctx context.Context
|
||||
svcCtx *svc.ServiceContext
|
||||
}
|
||||
|
||||
func NewRenderNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *RenderNotifyLogic {
|
||||
return &RenderNotifyLogic{
|
||||
Logger: logx.WithContext(ctx),
|
||||
ctx: ctx,
|
||||
svcCtx: svcCtx,
|
||||
}
|
||||
}
|
||||
|
||||
// 处理进入前逻辑w,r
|
||||
// func (l *RenderNotifyLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) {
|
||||
// }
|
||||
|
||||
// 处理逻辑后 w,r 如:重定向, resp 必须重新处理
|
||||
// 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) {
|
||||
// 返回值必须调用Set重新返回, resp可以空指针调用 resp.SetStatus(basic.CodeOK, data)
|
||||
// userinfo 传入值时, 一定不为null
|
||||
|
||||
return resp.SetStatus(basic.CodeOK)
|
||||
}
|
|
@ -6,13 +6,39 @@ import (
|
|||
)
|
||||
|
||||
type DataTransferReq struct {
|
||||
MsgType string `json:"msg_type"` //消息类型
|
||||
Message interface{} `json:"message"` //传递的消息
|
||||
T string `json:"t"` //消息类型
|
||||
D string `json:"d"` //传递的消息
|
||||
}
|
||||
|
||||
type DataTransferRsp struct {
|
||||
MsgType string `json:"msg_type"` //消息类型
|
||||
Message interface{} `json:"message"` //传递的消息
|
||||
T string `json:"t"` //消息类型
|
||||
D string `json:"d"` //传递的消息
|
||||
}
|
||||
|
||||
type RenderImageReqMsg struct {
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
}
|
||||
|
||||
type RenderImageRspMsg struct {
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type RenderNotifyReq struct {
|
||||
Sign string `json:"sign"`
|
||||
Time int64 `json:"time"`
|
||||
NotifyList []NotifyItem `json:"notify_list"`
|
||||
}
|
||||
|
||||
type NotifyItem struct {
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
type Request struct {
|
||||
|
|
25
server_api/product-template-tag.api
Normal file
25
server_api/product-template-tag.api
Normal file
|
@ -0,0 +1,25 @@
|
|||
syntax = "v1"
|
||||
|
||||
info (
|
||||
title: "产品模板标签服务"// TODO: add title
|
||||
desc: // TODO: add description
|
||||
author: ""
|
||||
email: ""
|
||||
)
|
||||
|
||||
import "basic.api"
|
||||
|
||||
service product-template-tag {
|
||||
//获取产品模板标签列表
|
||||
@handler GetProductTemplateTagsHandler
|
||||
get /api/product-template/get_product_template_tags(GetProductTemplateTagsReq) returns (response);
|
||||
}
|
||||
|
||||
//获取产品模板标签列表
|
||||
type GetProductTemplateTagsReq {
|
||||
Limit int `form:"limit"`
|
||||
}
|
||||
type GetProductTemplateTagsRsp {
|
||||
Tag string `json:"tag"`
|
||||
Cover string `json:"cover"`
|
||||
}
|
|
@ -12,14 +12,40 @@ service websocket {
|
|||
//websocket数据交互
|
||||
@handler DataTransferHandler
|
||||
get /api/websocket/data_transfer(DataTransferReq) returns (response);
|
||||
//渲染完了通知接口
|
||||
@handler RenderNotifyHandler
|
||||
post /api/websocket/render_notify(RenderNotifyReq) returns (response);
|
||||
}
|
||||
|
||||
//websocket数据交互
|
||||
type DataTransferReq {
|
||||
MsgType string `json:"msg_type"` //消息类型
|
||||
Message interface{} `json:"message"` //传递的消息
|
||||
T string `json:"t"` //消息类型
|
||||
D string `json:"d"` //传递的消息
|
||||
}
|
||||
type DataTransferRsp {
|
||||
MsgType string `json:"msg_type"` //消息类型
|
||||
Message interface{} `json:"message"` //传递的消息
|
||||
T string `json:"t"` //消息类型
|
||||
D string `json:"d"` //传递的消息
|
||||
}
|
||||
type RenderImageReqMsg { //websocket接受需要云渲染的图片
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
}
|
||||
type RenderImageRspMsg { //websocket发送渲染完的数据
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
//渲染完了通知接口
|
||||
type RenderNotifyReq {
|
||||
Sign string `json:"sign"`
|
||||
Time int64 `json:"time"`
|
||||
NotifyList []NotifyItem `json:"notify_list"`
|
||||
}
|
||||
type NotifyItem {
|
||||
ProductId int64 `json:"product_id"`
|
||||
SizeId int64 `json:"size_id"`
|
||||
TemplateId int64 `json:"template_id"`
|
||||
Source string `json:"source"`
|
||||
}
|
Loading…
Reference in New Issue
Block a user