SCHISANDRA/schisandra-album-cloud-microservices
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ complete mobile image upload

Browse Source
This commit is contained in:
landaiqing
2024-12-18 01:08:25 +08:00
parent 58777e4b58
commit 24df28e53f
15 changed files with 526 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
app/core/api/core.api
View File

@@ -125,6 +125,15 @@ type (
}
)
// 上传图片请求参数
type (
UploadRequest {
Image string `json:"image"`
AccessToken string `json:"access_token"`
userId string `json:"user_id"`
}
)
// 统一响应参数
type (
Response {
@@ -203,6 +212,9 @@ service core {
@handler messageWebsocket
get /message
@handler fileWebsocket
get /file
}
@server (
@@ -314,3 +326,18 @@ service core {
post /dislike (CommentDisLikeRequest) returns (Response)
}
@server (
group: upscale // 微服务分组
prefix: /api/auth/upscale // 微服务前缀
timeout: 10s // 超时时间
maxBytes: 10485760 // 最大请求大小
signature: false // 是否开启签名验证
middleware: SecurityHeadersMiddleware // 注册中间件
MaxConns: true // 是否开启最大连接数限制
Recover: true // 是否开启自动恢复
)
service core {
@handler uploadImage
post /upload (UploadRequest) returns (Response)
}

3
app/core/api/core.go
View File

@@ -10,7 +10,6 @@ import (
"schisandra-album-cloud-microservices/app/core/api/common/middleware"
"schisandra-album-cloud-microservices/app/core/api/internal/config"
"schisandra-album-cloud-microservices/app/core/api/internal/handler"
"schisandra-album-cloud-microservices/app/core/api/internal/logic/websocket"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
"schisandra-album-cloud-microservices/app/core/api/repository/idgenerator"
)
@@ -34,8 +33,6 @@ func main() {
handler.RegisterHandlers(server, ctx)
// init id generator
idgenerator.NewIDGenerator()
// init websocket handler
websocket.InitializeWebSocketHandler(ctx)
fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
server.Start()
}

2
app/core/api/etc/core.yaml
View File

@@ -7,7 +7,7 @@ Web:
Middlewares:
Log: true
Mysql:
DataSource: root:1611@tcp(127.0.0.1:3306)/schisandra-cloud-album?charset=utf8mb4&parseTime=True&loc=Local
DataSource: root:LDQ20020618xxx@tcp(1.95.0.111:3306)/schisandra-cloud-album?charset=utf8mb4&parseTime=True&loc=Local
MaxOpenConn: 10
MaxIdleConn: 5
Auth:

22
app/core/api/internal/handler/routes.go
View File

@@ -13,6 +13,7 @@ import (
oauth "schisandra-album-cloud-microservices/app/core/api/internal/handler/oauth"
sms "schisandra-album-cloud-microservices/app/core/api/internal/handler/sms"
token "schisandra-album-cloud-microservices/app/core/api/internal/handler/token"
upscale "schisandra-album-cloud-microservices/app/core/api/internal/handler/upscale"
user "schisandra-album-cloud-microservices/app/core/api/internal/handler/user"
websocket "schisandra-album-cloud-microservices/app/core/api/internal/handler/websocket"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
@@ -199,6 +200,22 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
rest.WithMaxBytes(1048576),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.SecurityHeadersMiddleware},
[]rest.Route{
{
Method: http.MethodPost,
Path: "/upload",
Handler: upscale.UploadImageHandler(serverCtx),
},
}...,
),
rest.WithPrefix("/api/auth/upscale"),
rest.WithTimeout(10000*time.Millisecond),
rest.WithMaxBytes(10485760),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.SecurityHeadersMiddleware},
@@ -233,6 +250,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
server.AddRoutes(
[]rest.Route{
{
Method: http.MethodGet,
Path: "/file",
Handler: websocket.FileWebsocketHandler(serverCtx),
},
{
Method: http.MethodGet,
Path: "/message",

35
app/core/api/internal/handler/upscale/upload_image_handler.go Normal file
View File

@@ -0,0 +1,35 @@
package upscale
import (
"github.com/zeromicro/go-zero/core/logx"
"net/http"
"github.com/zeromicro/go-zero/rest/httpx"
"schisandra-album-cloud-microservices/app/core/api/common/response"
"schisandra-album-cloud-microservices/app/core/api/internal/logic/upscale"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
"schisandra-album-cloud-microservices/app/core/api/internal/types"
)
func UploadImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.UploadRequest
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
l := upscale.NewUploadImageLogic(r.Context(), svcCtx)
resp, err := l.UploadImage(r, &req)
if err != nil {
logx.Error(err)
httpx.WriteJsonCtx(
r.Context(),
w,
http.StatusInternalServerError,
response.ErrorWithI18n(r.Context(), "system.error"))
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

15
app/core/api/internal/handler/websocket/file_websocket_handler.go Normal file
View File

@@ -0,0 +1,15 @@
package websocket
import (
"net/http"
"schisandra-album-cloud-microservices/app/core/api/internal/logic/websocket"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
)
func FileWebsocketHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
l := websocket.NewFileWebsocketLogic(r.Context(), svcCtx)
l.FileWebsocket(w, r)
}
}

54
app/core/api/internal/logic/upscale/upload_image_logic.go Normal file
View File

@@ -0,0 +1,54 @@
package upscale
import (
"context"
"encoding/json"
"net/http"
"schisandra-album-cloud-microservices/app/core/api/common/jwt"
"schisandra-album-cloud-microservices/app/core/api/common/response"
"schisandra-album-cloud-microservices/app/core/api/internal/logic/websocket"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
"schisandra-album-cloud-microservices/app/core/api/internal/types"
"github.com/zeromicro/go-zero/core/logx"
)
type UploadImageLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewUploadImageLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UploadImageLogic {
return &UploadImageLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *UploadImageLogic) UploadImage(r *http.Request, req *types.UploadRequest) (resp *types.Response, err error) {
token, ok := jwt.ParseAccessToken(l.svcCtx.Config.Auth.AccessSecret, req.AccessToken)
if !ok {
return response.ErrorWithI18n(l.ctx, "upload.uploadError"), nil
}
if token.UserID != req.UserId {
return response.ErrorWithI18n(l.ctx, "upload.uploadError"), nil
}
correct, err := l.svcCtx.CasbinEnforcer.Enforce(req.UserId, r.URL.Path, r.Method)
if err != nil || !correct {
return response.ErrorWithI18n(l.ctx, "upload.uploadError"), err
}
data, err := json.Marshal(response.SuccessWithData(req.Image))
if err != nil {
return nil, err
}
err = websocket.FileWebSocketHandler.SendMessageToClient(req.UserId, data)
if err != nil {
return nil, err
}
return response.Success(), nil
}

125
app/core/api/internal/logic/websocket/file_websocket_logic.go Normal file
View File

@@ -0,0 +1,125 @@
package websocket
import (
"context"
"fmt"
"github.com/lxzan/gws"
"net/http"
"schisandra-album-cloud-microservices/app/core/api/common/jwt"
"time"
"github.com/zeromicro/go-zero/core/logx"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
)
type FileWebsocketLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewFileWebsocketLogic(ctx context.Context, svcCtx *svc.ServiceContext) *FileWebsocketLogic {
return &FileWebsocketLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
type FileWebSocket struct {
ctx context.Context
gws.BuiltinEventHandler
sessions *gws.ConcurrentMap[string, *gws.Conn]
}
var FileWebSocketHandler = NewFileWebSocket()
func (l *FileWebsocketLogic) FileWebsocket(w http.ResponseWriter, r *http.Request) {
token := r.Header.Get("Sec-Websocket-Protocol")
accessToken, res := jwt.ParseAccessToken(l.svcCtx.Config.Auth.AccessSecret, token)
if !res {
return
}
upGrader := gws.NewUpgrader(FileWebSocketHandler, &gws.ServerOption{
HandshakeTimeout: 5 * time.Second, // 握手超时时间
ReadBufferSize: 1024, // 读缓冲区大小
ParallelEnabled: true, // 开启并行消息处理
Recovery: gws.Recovery, // 开启异常恢复
CheckUtf8Enabled: false, // 关闭UTF8校验
PermessageDeflate: gws.PermessageDeflate{
Enabled: true, // 开启压缩
},
Authorize: func(r *http.Request, session gws.SessionStorage) bool {
var clientId = r.URL.Query().Get("user_id")
if clientId == "" {
return false
}
if accessToken.UserID != clientId {
return false
}
session.Store("user_id", clientId)
return true
},
SubProtocols: []string{token},
})
socket, err := upGrader.Upgrade(w, r)
if err != nil {
panic(err)
}
go func() {
socket.ReadLoop()
}()
}
func NewFileWebSocket() *FileWebSocket {
return &FileWebSocket{
ctx: context.Background(),
sessions: gws.NewConcurrentMap[string, *gws.Conn](64, 128),
}
}
// OnOpen 连接建立
func (c *FileWebSocket) OnOpen(socket *gws.Conn) {
clientId := MustLoad[string](socket.Session(), "user_id")
c.sessions.Store(clientId, socket)
// 订阅该用户的频道
fmt.Printf("websocket client %s connected\n", clientId)
}
// OnClose 关闭连接
func (c *FileWebSocket) OnClose(socket *gws.Conn, err error) {
name := MustLoad[string](socket.Session(), "user_id")
sharding := c.sessions.GetSharding(name)
c.sessions.Delete(name)
sharding.Lock()
defer sharding.Unlock()
fmt.Printf("websocket closed, name=%s, msg=%s\n", name, err.Error())
}
// OnPing 处理客户端的Ping消息
func (c *FileWebSocket) OnPing(socket *gws.Conn, payload []byte) {
_ = socket.SetDeadline(time.Now().Add(PingInterval + HeartbeatWaitTimeout))
_ = socket.WritePong(payload)
}
// OnPong 处理客户端的Pong消息
func (c *FileWebSocket) OnPong(_ *gws.Conn, _ []byte) {}
// OnMessage 接受消息
func (c *FileWebSocket) OnMessage(socket *gws.Conn, message *gws.Message) {
defer message.Close()
clientId := MustLoad[string](socket.Session(), "user_id")
if conn, ok := c.sessions.Load(clientId); ok {
_ = conn.WriteMessage(gws.OpcodeText, message.Bytes())
}
// fmt.Printf("received message from client %s\n", message.Data)
}
// SendMessageToClient 向指定客户端发送消息
func (c *FileWebSocket) SendMessageToClient(clientId string, message []byte) error {
conn, ok := c.sessions.Load(clientId)
if ok {
return conn.WriteMessage(gws.OpcodeText, message)
}
return fmt.Errorf("client %s not found", clientId)
}

105
app/core/api/internal/logic/websocket/message_websocket_logic.go
View File

@@ -7,10 +7,8 @@ import (
"time"
"github.com/lxzan/gws"
"github.com/redis/go-redis/v9"
"github.com/zeromicro/go-zero/core/logx"
"schisandra-album-cloud-microservices/app/core/api/common/constant"
"schisandra-album-cloud-microservices/app/core/api/common/jwt"
"schisandra-album-cloud-microservices/app/core/api/internal/svc"
)
@@ -30,19 +28,19 @@ func NewMessageWebsocketLogic(ctx context.Context, svcCtx *svc.ServiceContext) *
}
type MessageWebSocket struct {
redis *redis.Client
ctx context.Context
ctx context.Context
gws.BuiltinEventHandler
sessions *gws.ConcurrentMap[string, *gws.Conn] // 使用内置的ConcurrentMap存储连接, 可以减少锁冲突
}
var MessageWebSocketHandler *MessageWebSocket
var MessageWebSocketHandler = NewMessageWebSocket()
// InitializeWebSocketHandler 初始化WebSocketHandler
func InitializeWebSocketHandler(svcCtx *svc.ServiceContext) {
MessageWebSocketHandler = NewMessageWebSocket(svcCtx)
}
func (l *MessageWebsocketLogic) MessageWebsocket(w http.ResponseWriter, r *http.Request) {
token := r.Header.Get("Sec-Websocket-Protocol")
accessToken, res := jwt.ParseAccessToken(l.svcCtx.Config.Auth.AccessSecret, token)
if !res {
return
}
upgrader := gws.NewUpgrader(MessageWebSocketHandler, &gws.ServerOption{
HandshakeTimeout: 5 * time.Second, // 握手超时时间
ReadBufferSize: 1024, // 读缓冲区大小
@@ -57,31 +55,34 @@ func (l *MessageWebsocketLogic) MessageWebsocket(w http.ResponseWriter, r *http.
if clientId == "" {
return false
}
token := r.URL.Query().Get("token")
if token == "" {
return false
}
accessToken, res := jwt.ParseAccessToken(l.svcCtx.Config.Auth.AccessSecret, token)
if !res || accessToken.UserID != clientId {
if accessToken.UserID != clientId {
return false
}
//token := r.URL.Query().Get("token")
//if token == "" {
// return false
//}
//accessToken, res := jwt.ParseAccessToken(l.svcCtx.Config.Auth.AccessSecret, token)
//if !res || accessToken.UserID != clientId {
// return false
//}
session.Store("user_id", clientId)
return true
},
SubProtocols: []string{token},
})
socket, err := upgrader.Upgrade(w, r)
if err != nil {
panic(err)
}
go func() {
socket.ReadLoop() // 此处阻塞会使请求上下文不能顺利被GC
socket.ReadLoop()
}()
}
// NewMessageWebSocket 创建WebSocket实例
func NewMessageWebSocket(svcCtx *svc.ServiceContext) *MessageWebSocket {
func NewMessageWebSocket() *MessageWebSocket {
return &MessageWebSocket{
redis: svcCtx.RedisClient,
ctx: context.Background(),
sessions: gws.NewConcurrentMap[string, *gws.Conn](64, 128),
}
@@ -92,7 +93,6 @@ func (c *MessageWebSocket) OnOpen(socket *gws.Conn) {
clientId := MustLoad[string](socket.Session(), "user_id")
c.sessions.Store(clientId, socket)
// 订阅该用户的频道
go c.subscribeUserChannel(clientId, c.redis)
fmt.Printf("websocket client %s connected\n", clientId)
}
@@ -124,70 +124,3 @@ func (c *MessageWebSocket) OnMessage(socket *gws.Conn, message *gws.Message) {
}
// fmt.Printf("received message from client %s\n", message.Data)
}
// SendMessageToClient 向指定客户端发送消息
func (c *MessageWebSocket) SendMessageToClient(clientId string, message []byte) error {
conn, ok := c.sessions.Load(clientId)
if ok {
return conn.WriteMessage(gws.OpcodeText, message)
}
return fmt.Errorf("client %s not found", clientId)
}
// SendMessageToUser 发送消息到指定用户的 Redis 频道
func (c *MessageWebSocket) SendMessageToUser(clientId string, message []byte, redis *redis.Client, ctx context.Context) error {
if _, ok := c.sessions.Load(clientId); ok {
return redis.Publish(ctx, clientId, message).Err()
} else {
return redis.LPush(ctx, constant.CommentOfflineMessagePrefix+clientId, message).Err()
}
}
// 订阅用户频道
func (c *MessageWebSocket) subscribeUserChannel(clientId string, redis *redis.Client) {
conn, ok := c.sessions.Load(clientId)
if !ok {
return
}
// 获取离线消息
messages, err := redis.LRange(c.ctx, constant.CommentOfflineMessagePrefix+clientId, 0, -1).Result()
if err != nil {
fmt.Printf("Error loading offline messages for user %s: %v\n", clientId, err)
return
}
// 逐条发送离线消息
for _, msg := range messages {
if writeErr := conn.WriteMessage(gws.OpcodeText, []byte(msg)); writeErr != nil {
fmt.Printf("Error writing offline message to user %s: %v\n", clientId, writeErr)
return
}
}
// 清空离线消息列表
if delErr := redis.Del(c.ctx, constant.CommentOfflineMessagePrefix+clientId); delErr.Err() != nil {
fmt.Printf("Error clearing offline messages for user %s: %v\n", clientId, delErr.Err())
return
}
pubsub := redis.Subscribe(c.ctx, clientId)
defer func() {
if closeErr := pubsub.Close(); closeErr != nil {
fmt.Printf("Error closing pubsub for user %s: %v\n", clientId, closeErr)
}
}()
for {
msg, waitErr := pubsub.ReceiveMessage(context.Background())
if waitErr != nil {
fmt.Printf("Error receiving message for user %s: %v\n", clientId, err)
return
}
if writeErr := conn.WriteMessage(gws.OpcodeText, []byte(msg.Payload)); writeErr != nil {
fmt.Printf("Error writing message to user %s: %v\n", clientId, writeErr)
return
}
}
}

6
app/core/api/internal/types/types.go
View File

@@ -112,3 +112,9 @@ type SmsSendRequest struct {
Angle int64 `json:"angle"`
Key string `json:"key"`
}
type UploadRequest struct {
Image string `json:"image"`
AccessToken string `json:"access_token"`
UserId string `json:"user_id"`
}

5
app/core/api/resources/language/active.en.toml
View File

@@ -25,4 +25,7 @@ smsSendSuccess = "sms send success！"
tooManyImages = "too many images！"
commentError = "comment error！"
LikeError = "like error！"
CancelLikeError = "cancel like error！"
CancelLikeError = "cancel like error！"
[upload]
uploadError = "upload error！"
uploadSuccess = "upload success！"

3
app/core/api/resources/language/active.zh.toml
View File

@@ -26,3 +26,6 @@ tooManyImages = "图片数量过多，请上传不超过3张！"
commentError = "评论失败！"
LikeError = "点赞失败！"
CancelLikeError = "取消点赞失败！"
[upload]
uploadError = "上传失败！"
uploadSuccess = "上传成功！"