mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-04-29 14:59:19 +08:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Gordon
|
9673f50518 | ||
|
withchao
|
5c31d12253 | ||
|
Gordon
|
bba662f404 | ||
|
charles-chenzz
|
7a3c3d7939 | ||
|
withchao
|
5fb9e946fc | ||
|
OpenIM Bot
|
40075de484 | ||
|
withchao
|
3e1b147160 |
+4
-1
@@ -132,7 +132,7 @@ api:
|
||||
# minio.signEndpoint is minio public network address
|
||||
object:
|
||||
enable: "minio"
|
||||
apiURL: "http://http://127.0.0.1:10002"
|
||||
apiURL: "http://127.0.0.1:10002"
|
||||
minio:
|
||||
bucket: "openim"
|
||||
endpoint: "http://172.28.0.1:10005"
|
||||
@@ -140,11 +140,13 @@ object:
|
||||
secretAccessKey: "openIM123"
|
||||
sessionToken: ''
|
||||
signEndpoint: "http://127.0.0.1:10005"
|
||||
publicRead: false
|
||||
cos:
|
||||
bucketURL: https://temp-1252357374.cos.ap-chengdu.myqcloud.com
|
||||
secretID: ''
|
||||
secretKey: ''
|
||||
sessionToken: ''
|
||||
publicRead: false
|
||||
oss:
|
||||
endpoint: "https://oss-cn-chengdu.aliyuncs.com"
|
||||
bucket: "demo-9999999"
|
||||
@@ -152,6 +154,7 @@ object:
|
||||
accessKeyID: ''
|
||||
accessKeySecret: ''
|
||||
sessionToken: ''
|
||||
publicRead: false
|
||||
|
||||
|
||||
###################### RPC Port Configuration ######################
|
||||
|
||||
@@ -140,11 +140,13 @@ object:
|
||||
secretAccessKey: "${MINIO_SECRET_KEY}"
|
||||
sessionToken: ${MINIO_SESSION_TOKEN}
|
||||
signEndpoint: "${MINIO_SIGN_ENDPOINT}"
|
||||
publicRead: ${MINIO_PUBLIC_READ}
|
||||
cos:
|
||||
bucketURL: ${COS_BUCKET_URL}
|
||||
secretID: ${COS_SECRET_ID}
|
||||
secretKey: ${COS_SECRET_KEY}
|
||||
sessionToken: ${COS_SESSION_TOKEN}
|
||||
publicRead: ${COS_PUBLIC_READ}
|
||||
oss:
|
||||
endpoint: "${OSS_ENDPOINT}"
|
||||
bucket: "${OSS_BUCKET}"
|
||||
@@ -152,7 +154,7 @@ object:
|
||||
accessKeyID: ${OSS_ACCESS_KEY_ID}
|
||||
accessKeySecret: ${OSS_ACCESS_KEY_SECRET}
|
||||
sessionToken: ${OSS_SESSION_TOKEN}
|
||||
|
||||
publicRead: ${OSS_PUBLIC_READ}
|
||||
|
||||
###################### RPC Port Configuration ######################
|
||||
# RPC service ports
|
||||
|
||||
@@ -37,7 +37,7 @@ require github.com/google/uuid v1.3.1
|
||||
|
||||
require (
|
||||
github.com/IBM/sarama v1.41.1
|
||||
github.com/OpenIMSDK/protocol v0.0.18
|
||||
github.com/OpenIMSDK/protocol v0.0.23
|
||||
github.com/OpenIMSDK/tools v0.0.14
|
||||
github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
|
||||
github.com/go-redis/redis v6.15.9+incompatible
|
||||
|
||||
@@ -19,8 +19,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/IBM/sarama v1.41.1 h1:B4/TdHce/8Ipza+qrLIeNJ9D1AOxZVp/3uDv6H/dp2M=
|
||||
github.com/IBM/sarama v1.41.1/go.mod h1:JFCPURVskaipJdKRFkiE/OZqQHw7jqliaJmRwXCmSSw=
|
||||
github.com/OpenIMSDK/protocol v0.0.18 h1:hXukFiDMLZx7s+hDCQePIK9ABiHyNlobNL4MppvOuMY=
|
||||
github.com/OpenIMSDK/protocol v0.0.18/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
|
||||
github.com/OpenIMSDK/protocol v0.0.23 h1:L545aRQez6Ro+AaJB1Z6Mz7ojnDtp41WqASxYveCkcE=
|
||||
github.com/OpenIMSDK/protocol v0.0.23/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
|
||||
github.com/OpenIMSDK/tools v0.0.14 h1:WLof/+WxyPyRST+QkoTKubYCiV73uCLiL8pgnpH/yKQ=
|
||||
github.com/OpenIMSDK/tools v0.0.14/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
|
||||
github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
|
||||
|
||||
@@ -11,4 +11,5 @@ use (
|
||||
./tools/versionchecker
|
||||
./tools/yamlfmt
|
||||
./tools/component
|
||||
./tools/url2im
|
||||
)
|
||||
|
||||
@@ -44,3 +44,7 @@ func (o *ConversationApi) GetConversations(c *gin.Context) {
|
||||
func (o *ConversationApi) SetConversations(c *gin.Context) {
|
||||
a2r.Call(conversation.ConversationClient.SetConversations, o.Client, c)
|
||||
}
|
||||
|
||||
func (o *ConversationApi) GetConversationOfflinePushUserIDs(c *gin.Context) {
|
||||
a2r.Call(conversation.ConversationClient.GetConversationOfflinePushUserIDs, o.Client, c)
|
||||
}
|
||||
|
||||
+3
-2
@@ -58,7 +58,7 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
|
||||
options := make(map[string]bool, 5)
|
||||
switch params.ContentType {
|
||||
case constant.Text:
|
||||
newContent = params.Content["text"].(string)
|
||||
fallthrough
|
||||
case constant.Picture:
|
||||
fallthrough
|
||||
case constant.Custom:
|
||||
@@ -100,6 +100,7 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
|
||||
ContentType: params.ContentType,
|
||||
Content: []byte(newContent),
|
||||
CreateTime: utils.GetCurrentTimestampByMill(),
|
||||
SendTime: params.SendTime,
|
||||
Options: options,
|
||||
OfflinePushInfo: params.OfflinePushInfo,
|
||||
},
|
||||
@@ -207,7 +208,6 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
|
||||
apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
|
||||
return
|
||||
}
|
||||
log.ZInfo(c, "SendMessage", "req", req)
|
||||
if !authverify.IsAppManagerUid(c) {
|
||||
apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message"))
|
||||
return
|
||||
@@ -224,6 +224,7 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
|
||||
respPb, err := m.Client.SendMsg(c, sendMsgReq)
|
||||
if err != nil {
|
||||
status = constant.MsgSendFailed
|
||||
log.ZError(c, "send message err", err)
|
||||
apiresp.GinError(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -156,6 +156,11 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
|
||||
thirdGroup.POST("/fcm_update_token", t.FcmUpdateToken)
|
||||
thirdGroup.POST("/set_app_badge", t.SetAppBadge)
|
||||
|
||||
logs := thirdGroup.Group("/logs")
|
||||
logs.POST("/upload", t.UploadLogs)
|
||||
logs.POST("/delete", t.DeleteLogs)
|
||||
logs.POST("/search", t.SearchLogs)
|
||||
|
||||
objectGroup := r.Group("/object", ParseToken)
|
||||
|
||||
objectGroup.POST("/part_limit", t.PartLimit)
|
||||
@@ -198,6 +203,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
|
||||
conversationGroup.POST("/get_conversation", c.GetConversation)
|
||||
conversationGroup.POST("/get_conversations", c.GetConversations)
|
||||
conversationGroup.POST("/set_conversations", c.SetConversations)
|
||||
conversationGroup.POST("/get_conversation_offline_push_user_ids", c.GetConversationOfflinePushUserIDs)
|
||||
}
|
||||
|
||||
statisticsGroup := r.Group("/statistics", ParseToken)
|
||||
|
||||
@@ -105,3 +105,16 @@ func (o *ThirdApi) ObjectRedirect(c *gin.Context) {
|
||||
}
|
||||
c.Redirect(http.StatusFound, resp.Url)
|
||||
}
|
||||
|
||||
// #################### logs ####################
|
||||
func (o *ThirdApi) UploadLogs(c *gin.Context) {
|
||||
a2r.Call(third.ThirdClient.UploadLogs, o.Client, c)
|
||||
}
|
||||
|
||||
func (o *ThirdApi) DeleteLogs(c *gin.Context) {
|
||||
a2r.Call(third.ThirdClient.DeleteLogs, o.Client, c)
|
||||
}
|
||||
|
||||
func (o *ThirdApi) SearchLogs(c *gin.Context) {
|
||||
a2r.Call(third.ThirdClient.SearchLogs, o.Client, c)
|
||||
}
|
||||
|
||||
@@ -59,7 +59,7 @@ const (
|
||||
PongMessage = 10
|
||||
)
|
||||
|
||||
type PongHandler func(string) error
|
||||
type PingPongHandler func(string) error
|
||||
|
||||
type Client struct {
|
||||
w *sync.Mutex
|
||||
@@ -107,8 +107,12 @@ func (c *Client) ResetClient(
|
||||
c.token = token
|
||||
}
|
||||
|
||||
func (c *Client) pongHandler(_ string) error {
|
||||
func (c *Client) pingHandler(_ string) error {
|
||||
c.conn.SetReadDeadline(pongWait)
|
||||
err := c.writePongMsg()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -122,10 +126,11 @@ func (c *Client) readMessage() {
|
||||
}()
|
||||
c.conn.SetReadLimit(maxMessageSize)
|
||||
_ = c.conn.SetReadDeadline(pongWait)
|
||||
c.conn.SetPongHandler(c.pongHandler)
|
||||
c.conn.SetPingHandler(c.pingHandler)
|
||||
for {
|
||||
messageType, message, returnErr := c.conn.ReadMessage()
|
||||
if returnErr != nil {
|
||||
log.ZWarn(c.ctx, "readMessage", returnErr, "messageType", messageType)
|
||||
c.closedErr = returnErr
|
||||
return
|
||||
}
|
||||
|
||||
@@ -41,7 +41,8 @@ type LongConn interface {
|
||||
SetConnNil()
|
||||
// SetReadLimit sets the maximum size for a message read from the peer.bytes
|
||||
SetReadLimit(limit int64)
|
||||
SetPongHandler(handler PongHandler)
|
||||
SetPongHandler(handler PingPongHandler)
|
||||
SetPingHandler(handler PingPongHandler)
|
||||
// GenerateLongConn Check the connection of the current and when it was sent are the same
|
||||
GenerateLongConn(w http.ResponseWriter, r *http.Request) error
|
||||
}
|
||||
@@ -116,9 +117,12 @@ func (d *GWebSocket) SetReadLimit(limit int64) {
|
||||
d.conn.SetReadLimit(limit)
|
||||
}
|
||||
|
||||
func (d *GWebSocket) SetPongHandler(handler PongHandler) {
|
||||
func (d *GWebSocket) SetPongHandler(handler PingPongHandler) {
|
||||
d.conn.SetPongHandler(handler)
|
||||
}
|
||||
func (d *GWebSocket) SetPingHandler(handler PingPongHandler) {
|
||||
d.conn.SetPingHandler(handler)
|
||||
}
|
||||
|
||||
//func (d *GWebSocket) CheckSendConnDiffNow() bool {
|
||||
// return d.conn == d.sendConn
|
||||
|
||||
@@ -60,7 +60,7 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(
|
||||
log.ZError(ctx, "msg_transfer Unmarshal msg err", err)
|
||||
return
|
||||
}
|
||||
return
|
||||
|
||||
log.ZDebug(ctx, "handleChatWs2Mysql", "msg", msgFromMQ.MsgData)
|
||||
// Control whether to store history messages (mysql)
|
||||
isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
|
||||
|
||||
@@ -19,6 +19,8 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/conversation"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
@@ -117,7 +119,6 @@ func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.Msg
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -234,15 +235,23 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
err = p.offlinePushMsg(ctx, groupID, msg, offlinePushUserIDs)
|
||||
resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
|
||||
ctx,
|
||||
&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
|
||||
)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
|
||||
return err
|
||||
}
|
||||
_, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
|
||||
if err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
|
||||
return err
|
||||
if len(resp.UserIDs) > 0 {
|
||||
err = p.offlinePushMsg(ctx, groupID, msg, resp.UserIDs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
|
||||
return err
|
||||
}
|
||||
if _, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(resp.UserIDs, WebAndPcBackgroundUserIDs)); err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -300,3 +300,30 @@ func (c *conversationServer) GetConversationsByConversationID(
|
||||
}
|
||||
return &pbconversation.GetConversationsByConversationIDResp{Conversations: convert.ConversationsDB2Pb(conversations)}, nil
|
||||
}
|
||||
|
||||
func (c *conversationServer) GetConversationOfflinePushUserIDs(
|
||||
ctx context.Context,
|
||||
req *pbconversation.GetConversationOfflinePushUserIDsReq,
|
||||
) (*pbconversation.GetConversationOfflinePushUserIDsResp, error) {
|
||||
if req.ConversationID == "" {
|
||||
return nil, errs.ErrArgs.Wrap("conversationID is empty")
|
||||
}
|
||||
if len(req.UserIDs) == 0 {
|
||||
return &pbconversation.GetConversationOfflinePushUserIDsResp{}, nil
|
||||
}
|
||||
userIDs, err := c.conversationDatabase.GetConversationNotReceiveMessageUserIDs(ctx, req.ConversationID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(userIDs) == 0 {
|
||||
return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: req.UserIDs}, nil
|
||||
}
|
||||
userIDSet := make(map[string]struct{})
|
||||
for _, userID := range req.UserIDs {
|
||||
userIDSet[userID] = struct{}{}
|
||||
}
|
||||
for _, userID := range userIDs {
|
||||
delete(userIDSet, userID)
|
||||
}
|
||||
return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: utils.Keys(userIDSet)}, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
package third
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/OpenIMSDK/protocol/third"
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
utils2 "github.com/OpenIMSDK/tools/utils"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
)
|
||||
|
||||
func genLogID() string {
|
||||
const dataLen = 10
|
||||
data := make([]byte, dataLen)
|
||||
rand.Read(data)
|
||||
chars := []byte("0123456789")
|
||||
for i := 0; i < len(data); i++ {
|
||||
if i == 0 {
|
||||
data[i] = chars[1:][data[i]%9]
|
||||
} else {
|
||||
data[i] = chars[data[i]%10]
|
||||
}
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
func (t *thirdServer) UploadLogs(ctx context.Context, req *third.UploadLogsReq) (*third.UploadLogsResp, error) {
|
||||
var DBlogs []*relationtb.Log
|
||||
userID := ctx.Value(constant.OpUserID).(string)
|
||||
platform := constant.PlatformID2Name[int(req.Platform)]
|
||||
for _, fileURL := range req.FileURLs {
|
||||
log := relationtb.Log{
|
||||
Version: req.Version,
|
||||
SystemType: req.SystemType,
|
||||
Platform: platform,
|
||||
UserID: userID,
|
||||
CreateTime: time.Now(),
|
||||
Url: fileURL.URL,
|
||||
FileName: fileURL.Filename,
|
||||
}
|
||||
for i := 0; i < 20; i++ {
|
||||
id := genLogID()
|
||||
logs, err := t.thirdDatabase.GetLogs(ctx, []string{id}, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(logs) == 0 {
|
||||
log.LogID = id
|
||||
break
|
||||
}
|
||||
}
|
||||
if log.LogID == "" {
|
||||
return nil, errs.ErrData.Wrap("Log id gen error")
|
||||
}
|
||||
DBlogs = append(DBlogs, &log)
|
||||
}
|
||||
err := t.thirdDatabase.UploadLogs(ctx, DBlogs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &third.UploadLogsResp{}, nil
|
||||
}
|
||||
|
||||
func (t *thirdServer) DeleteLogs(ctx context.Context, req *third.DeleteLogsReq) (*third.DeleteLogsResp, error) {
|
||||
|
||||
if err := authverify.CheckAdmin(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
userID := ""
|
||||
logs, err := t.thirdDatabase.GetLogs(ctx, req.LogIDs, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var logIDs []string
|
||||
for _, log := range logs {
|
||||
logIDs = append(logIDs, log.LogID)
|
||||
}
|
||||
if ids := utils2.Single(req.LogIDs, logIDs); len(ids) > 0 {
|
||||
return nil, errs.ErrRecordNotFound.Wrap(fmt.Sprintf("logIDs not found%#v", ids))
|
||||
}
|
||||
err = t.thirdDatabase.DeleteLogs(ctx, req.LogIDs, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &third.DeleteLogsResp{}, nil
|
||||
}
|
||||
|
||||
func dbToPbLogInfos(logs []*relationtb.Log) []*third.LogInfo {
|
||||
db2pbForLogInfo := func(log *relationtb.Log) *third.LogInfo {
|
||||
return &third.LogInfo{
|
||||
Filename: log.FileName,
|
||||
UserID: log.UserID,
|
||||
Platform: utils.StringToInt32(log.Platform),
|
||||
Url: log.Url,
|
||||
CreateTime: log.CreateTime.UnixMilli(),
|
||||
LogID: log.LogID,
|
||||
SystemType: log.SystemType,
|
||||
Version: log.Version,
|
||||
Ex: log.Ex,
|
||||
}
|
||||
}
|
||||
return utils.Slice(logs, db2pbForLogInfo)
|
||||
}
|
||||
|
||||
func (t *thirdServer) SearchLogs(ctx context.Context, req *third.SearchLogsReq) (*third.SearchLogsResp, error) {
|
||||
if err := authverify.CheckAdmin(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var (
|
||||
resp third.SearchLogsResp
|
||||
userIDs []string
|
||||
)
|
||||
if req.StartTime > req.EndTime {
|
||||
return nil, errs.ErrArgs.Wrap("startTime>endTime")
|
||||
}
|
||||
total, logs, err := t.thirdDatabase.SearchLogs(ctx, req.Keyword, time.UnixMilli(req.StartTime), time.UnixMilli(req.EndTime), req.Pagination.PageNumber, req.Pagination.ShowNumber)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pbLogs := dbToPbLogInfos(logs)
|
||||
for _, log := range logs {
|
||||
userIDs = append(userIDs, log.UserID)
|
||||
}
|
||||
users, err := t.thirdDatabase.FindUsers(ctx, userIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
IDtoName := make(map[string]string)
|
||||
for _, user := range users {
|
||||
IDtoName[user.UserID] = user.Nickname
|
||||
}
|
||||
for _, pbLog := range pbLogs {
|
||||
pbLog.Nickname = IDtoName[pbLog.UserID]
|
||||
}
|
||||
resp.LogsInfos = pbLogs
|
||||
resp.Total = total
|
||||
return &resp, nil
|
||||
}
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
)
|
||||
|
||||
@@ -79,7 +80,7 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
|
||||
}
|
||||
third.RegisterThirdServer(server, &thirdServer{
|
||||
apiURL: apiURL,
|
||||
thirdDatabase: controller.NewThirdDatabase(cache.NewMsgCacheModel(rdb)),
|
||||
thirdDatabase: controller.NewThirdDatabase(cache.NewMsgCacheModel(rdb), db),
|
||||
userRpcClient: rpcclient.NewUserRpcClient(client),
|
||||
s3dataBase: controller.NewS3Database(o, relation.NewObjectInfo(db)),
|
||||
defaultExpire: time.Hour * 24 * 7,
|
||||
|
||||
@@ -29,6 +29,7 @@ type SendMsg struct {
|
||||
SessionType int32 `json:"sessionType" binding:"required"`
|
||||
IsOnlineOnly bool `json:"isOnlineOnly"`
|
||||
NotOfflinePush bool `json:"notOfflinePush"`
|
||||
SendTime int64 `json:"sendTime"`
|
||||
OfflinePushInfo *sdkws.OfflinePushInfo `json:"offlinePushInfo"`
|
||||
}
|
||||
|
||||
|
||||
@@ -128,12 +128,14 @@ type configStruct struct {
|
||||
SecretAccessKey string `yaml:"secretAccessKey"`
|
||||
SessionToken string `yaml:"sessionToken"`
|
||||
SignEndpoint string `yaml:"signEndpoint"`
|
||||
PublicRead bool `yaml:"publicRead"`
|
||||
} `yaml:"minio"`
|
||||
Cos struct {
|
||||
BucketURL string `yaml:"bucketURL"`
|
||||
SecretID string `yaml:"secretID"`
|
||||
SecretKey string `yaml:"secretKey"`
|
||||
SessionToken string `yaml:"sessionToken"`
|
||||
PublicRead bool `yaml:"publicRead"`
|
||||
} `yaml:"cos"`
|
||||
Oss struct {
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
@@ -142,6 +144,7 @@ type configStruct struct {
|
||||
AccessKeyID string `yaml:"accessKeyID"`
|
||||
AccessKeySecret string `yaml:"accessKeySecret"`
|
||||
SessionToken string `yaml:"sessionToken"`
|
||||
PublicRead bool `yaml:"publicRead"`
|
||||
} `yaml:"oss"`
|
||||
} `yaml:"object"`
|
||||
|
||||
|
||||
Vendored
+33
-13
@@ -38,6 +38,7 @@ const (
|
||||
recvMsgOptKey = "RECV_MSG_OPT:"
|
||||
superGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
|
||||
conversationNotReceiveMessageUserIDsKey = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
|
||||
|
||||
conversationExpireTime = time.Second * 60 * 60 * 12
|
||||
)
|
||||
@@ -58,11 +59,8 @@ type ConversationCache interface {
|
||||
DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache
|
||||
// get one conversation from msgCache
|
||||
GetConversations(
|
||||
ctx context.Context,
|
||||
ownerUserID string,
|
||||
conversationIDs []string,
|
||||
) ([]*relationtb.ConversationModel, error)
|
||||
GetConversations(ctx context.Context, ownerUserID string,
|
||||
conversationIDs []string) ([]*relationtb.ConversationModel, error)
|
||||
// get one user's all conversations from msgCache
|
||||
GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.ConversationModel, error)
|
||||
// get user conversation recv msg from msgCache
|
||||
@@ -78,11 +76,11 @@ type ConversationCache interface {
|
||||
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
|
||||
DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
|
||||
GetConversationsByConversationID(
|
||||
ctx context.Context,
|
||||
conversationIDs []string,
|
||||
) ([]*relationtb.ConversationModel, error)
|
||||
GetConversationsByConversationID(ctx context.Context,
|
||||
conversationIDs []string) ([]*relationtb.ConversationModel, error)
|
||||
DelConversationByConversationID(conversationIDs ...string) ConversationCache
|
||||
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
|
||||
DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
|
||||
}
|
||||
|
||||
func NewConversationRedis(
|
||||
@@ -153,6 +151,10 @@ func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conve
|
||||
return conversationHasReadSeqKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
|
||||
return conversationNotReceiveMessageUserIDsKey + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
|
||||
return getCache(
|
||||
ctx,
|
||||
@@ -411,10 +413,8 @@ func (c *ConversationRedisCache) GetUserAllHasReadSeqs(
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(
|
||||
ownerUserID string,
|
||||
conversationIDs ...string,
|
||||
) ConversationCache {
|
||||
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string,
|
||||
conversationIDs ...string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationHasReadSeqKey(ownerUserID, conversationID))
|
||||
@@ -432,3 +432,23 @@ func (c *ConversationRedisCache) GetConversationsByConversationID(
|
||||
func (c *ConversationRedisCache) DelConversationByConversationID(conversationIDs ...string) ConversationCache {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
return getCache(
|
||||
ctx,
|
||||
c.rcClient,
|
||||
c.getConversationNotReceiveMessageUserIDsKey(conversationID),
|
||||
c.expireTime,
|
||||
func(ctx context.Context) ([]string, error) {
|
||||
return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationNotReceiveMessageUserIDsKey(conversationID))
|
||||
}
|
||||
return cache
|
||||
}
|
||||
|
||||
@@ -53,6 +53,7 @@ type ConversationDatabase interface {
|
||||
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
|
||||
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error)
|
||||
GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.ConversationModel, error)
|
||||
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
|
||||
}
|
||||
|
||||
func NewConversationDatabase(conversation relationtb.ConversationModelInterface, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
|
||||
@@ -88,6 +89,9 @@ func (c *conversationDatabase) SetUsersConversationFiledTx(ctx context.Context,
|
||||
cache = cache.DelUserAllHasReadSeqs(userID, conversation.ConversationID)
|
||||
}
|
||||
}
|
||||
if _, ok := filedMap["recv_msg_opt"]; ok {
|
||||
cache = cache.DelConversationNotReceiveMessageUserIDs(conversation.ConversationID)
|
||||
}
|
||||
}
|
||||
NotUserIDs := utils.DifferenceString(haveUserIDs, userIDs)
|
||||
log.ZDebug(ctx, "SetUsersConversationFiledTx", "NotUserIDs", NotUserIDs, "haveUserIDs", haveUserIDs, "userIDs", userIDs)
|
||||
@@ -121,7 +125,12 @@ func (c *conversationDatabase) UpdateUsersConversationFiled(ctx context.Context,
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.cache.DelUsersConversation(conversationID, userIDs...).ExecDel(ctx)
|
||||
cache := c.cache.NewCache()
|
||||
cache = cache.DelUsersConversation(conversationID, userIDs...)
|
||||
if _, ok := args["recv_msg_opt"]; ok {
|
||||
cache = cache.DelConversationNotReceiveMessageUserIDs(conversationID)
|
||||
}
|
||||
return cache.ExecDel(ctx)
|
||||
}
|
||||
|
||||
func (c *conversationDatabase) CreateConversation(ctx context.Context, conversations []*relationtb.ConversationModel) error {
|
||||
@@ -132,6 +141,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat
|
||||
cache := c.cache.NewCache()
|
||||
for _, conversation := range conversations {
|
||||
cache = cache.DelConversations(conversation.OwnerUserID, conversation.ConversationID)
|
||||
cache = cache.DelConversationNotReceiveMessageUserIDs(conversation.ConversationID)
|
||||
userIDs = append(userIDs, conversation.OwnerUserID)
|
||||
}
|
||||
return cache.DelConversationIDs(userIDs...).DelUserConversationIDsHash(userIDs...).ExecDel(ctx)
|
||||
@@ -224,7 +234,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID)
|
||||
cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID).DelConversationNotReceiveMessageUserIDs(utils.Slice(notExistConversations, func(e *relationtb.ConversationModel) string { return e.ConversationID })...)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
@@ -250,7 +260,7 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context,
|
||||
for _, v := range notExistUserIDs {
|
||||
conversation := relationtb.ConversationModel{ConversationType: constant.SuperGroupChatType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID}
|
||||
conversations = append(conversations, &conversation)
|
||||
cache = cache.DelConversations(v, conversationID)
|
||||
cache = cache.DelConversations(v, conversationID).DelConversationNotReceiveMessageUserIDs(conversationID)
|
||||
}
|
||||
cache = cache.DelConversationIDs(notExistUserIDs...).DelUserConversationIDsHash(notExistUserIDs...)
|
||||
if len(conversations) > 0 {
|
||||
@@ -296,3 +306,7 @@ func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Cont
|
||||
func (c *conversationDatabase) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.ConversationModel, error) {
|
||||
return c.conversationDB.GetConversationIDsNeedDestruct(ctx)
|
||||
}
|
||||
|
||||
func (c *conversationDatabase) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
return c.cache.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
|
||||
}
|
||||
|
||||
@@ -16,21 +16,60 @@ package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
dbimpl "github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
)
|
||||
|
||||
type ThirdDatabase interface {
|
||||
FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error
|
||||
SetAppBadge(ctx context.Context, userID string, value int) error
|
||||
//about log for debug
|
||||
UploadLogs(ctx context.Context, logs []*relation.Log) error
|
||||
DeleteLogs(ctx context.Context, logID []string, userID string) error
|
||||
SearchLogs(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relation.Log, error)
|
||||
GetLogs(ctx context.Context, LogIDs []string, userID string) ([]*relation.Log, error)
|
||||
FindUsers(ctx context.Context, userIDs []string) ([]*relation.UserModel, error)
|
||||
}
|
||||
|
||||
type thirdDatabase struct {
|
||||
cache cache.MsgModel
|
||||
cache cache.MsgModel
|
||||
logdb relation.LogInterface
|
||||
userdb relation.UserModelInterface
|
||||
}
|
||||
|
||||
func NewThirdDatabase(cache cache.MsgModel) ThirdDatabase {
|
||||
return &thirdDatabase{cache: cache}
|
||||
// FindUsers implements ThirdDatabase.
|
||||
func (t *thirdDatabase) FindUsers(ctx context.Context, userIDs []string) ([]*relation.UserModel, error) {
|
||||
return t.userdb.Find(ctx, userIDs)
|
||||
}
|
||||
|
||||
// DeleteLogs implements ThirdDatabase.
|
||||
func (t *thirdDatabase) DeleteLogs(ctx context.Context, logID []string, userID string) error {
|
||||
return t.logdb.Delete(ctx, logID, userID)
|
||||
}
|
||||
|
||||
// GetLogs implements ThirdDatabase.
|
||||
func (t *thirdDatabase) GetLogs(ctx context.Context, LogIDs []string, userID string) ([]*relation.Log, error) {
|
||||
return t.logdb.Get(ctx, LogIDs, userID)
|
||||
}
|
||||
|
||||
// SearchLogs implements ThirdDatabase.
|
||||
func (t *thirdDatabase) SearchLogs(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relation.Log, error) {
|
||||
return t.logdb.Search(ctx, keyword, start, end, pageNumber, showNumber)
|
||||
|
||||
}
|
||||
|
||||
// UploadLogs implements ThirdDatabase.
|
||||
func (t *thirdDatabase) UploadLogs(ctx context.Context, logs []*relation.Log) error {
|
||||
return t.logdb.Create(ctx, logs)
|
||||
}
|
||||
|
||||
func NewThirdDatabase(cache cache.MsgModel, db *gorm.DB) ThirdDatabase {
|
||||
return &thirdDatabase{cache: cache, logdb: dbimpl.NewLogGorm(db), userdb: dbimpl.NewUserGorm(db)}
|
||||
}
|
||||
|
||||
func (t *thirdDatabase) FcmUpdateToken(
|
||||
|
||||
@@ -17,6 +17,8 @@ package relation
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
@@ -214,3 +216,24 @@ func (c *ConversationGorm) GetConversationIDsNeedDestruct(
|
||||
"",
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) GetConversationRecvMsgOpt(ctx context.Context, userID string, conversationID string) (int32, error) {
|
||||
var recvMsgOpt int32
|
||||
return recvMsgOpt, errs.Wrap(
|
||||
c.db(ctx).
|
||||
Model(&relation.ConversationModel{}).
|
||||
Where("conversation_id = ? and owner_user_id in ?", conversationID, userID).
|
||||
Pluck("recv_msg_opt", &recvMsgOpt).
|
||||
Error,
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
var userIDs []string
|
||||
return userIDs, errs.Wrap(
|
||||
c.db(ctx).
|
||||
Model(&relation.ConversationModel{}).
|
||||
Where("conversation_id = ? and recv_msg_opt <> ?", conversationID, constant.ReceiveMessage).
|
||||
Pluck("owner_user_id", &userIDs).Error,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
package relation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"github.com/OpenIMSDK/tools/ormutil"
|
||||
"gorm.io/gorm"
|
||||
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
)
|
||||
|
||||
type LogGorm struct {
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
func (l *LogGorm) Create(ctx context.Context, log []*relationtb.Log) error {
|
||||
return errs.Wrap(l.db.WithContext(ctx).Create(log).Error)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relationtb.Log, error) {
|
||||
db := l.db.WithContext(ctx).Where("create_time >= ?", start)
|
||||
if end.UnixMilli() != 0 {
|
||||
db = l.db.WithContext(ctx).Where("create_time <= ?", end)
|
||||
}
|
||||
return ormutil.GormSearch[relationtb.Log](db, []string{"user_id"}, keyword, pageNumber, showNumber)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Delete(ctx context.Context, logIDs []string, userID string) error {
|
||||
if userID == "" {
|
||||
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Delete(&relationtb.Log{}).Error)
|
||||
}
|
||||
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Delete(&relationtb.Log{}).Error)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Get(ctx context.Context, logIDs []string, userID string) ([]*relationtb.Log, error) {
|
||||
var logs []*relationtb.Log
|
||||
if userID == "" {
|
||||
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Find(&logs).Error)
|
||||
}
|
||||
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Find(&logs).Error)
|
||||
}
|
||||
func NewLogGorm(db *gorm.DB) relationtb.LogInterface {
|
||||
db.AutoMigrate(&relationtb.Log{})
|
||||
return &LogGorm{db: db}
|
||||
}
|
||||
@@ -288,7 +288,7 @@ func (c *Cos) AccessURL(ctx context.Context, name string, expire time.Duration,
|
||||
style = append(style, "format/"+opt.Image.Format)
|
||||
}
|
||||
if len(style) > 0 {
|
||||
imageMogr = "&imageMogr2/thumbnail/" + strings.Join(style, "/") + "/ignore-error/1"
|
||||
imageMogr = "imageMogr2/thumbnail/" + strings.Join(style, "/") + "/ignore-error/1"
|
||||
}
|
||||
}
|
||||
if opt.ContentType != "" {
|
||||
@@ -306,13 +306,23 @@ func (c *Cos) AccessURL(ctx context.Context, name string, expire time.Duration,
|
||||
} else if expire < time.Second {
|
||||
expire = time.Second
|
||||
}
|
||||
rawURL, err := c.client.Object.GetPresignedURL(ctx, http.MethodGet, name, c.credential.SecretID, c.credential.SecretKey, expire, &option)
|
||||
rawURL, err := c.getPresignedURL(ctx, name, expire, &option)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
urlStr := rawURL.String()
|
||||
if imageMogr != "" {
|
||||
urlStr += imageMogr
|
||||
if rawURL.RawQuery == "" {
|
||||
rawURL.RawQuery = imageMogr
|
||||
} else {
|
||||
rawURL.RawQuery = rawURL.RawQuery + "&" + imageMogr
|
||||
}
|
||||
}
|
||||
return urlStr, nil
|
||||
return rawURL.String(), nil
|
||||
}
|
||||
|
||||
func (c *Cos) getPresignedURL(ctx context.Context, name string, expire time.Duration, opt *cos.PresignedURLOptions) (*url.URL, error) {
|
||||
if !config.Config.Object.Cos.PublicRead {
|
||||
return c.client.Object.GetPresignedURL(ctx, http.MethodGet, name, c.credential.SecretID, c.credential.SecretKey, expire, opt)
|
||||
}
|
||||
return c.client.Object.GetObjectURL(name), nil
|
||||
}
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package cos // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/cos"
|
||||
@@ -0,0 +1,13 @@
|
||||
package cos
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/url"
|
||||
_ "unsafe"
|
||||
|
||||
"github.com/tencentyun/cos-go-sdk-v5"
|
||||
)
|
||||
|
||||
//go:linkname newRequest github.com/tencentyun/cos-go-sdk-v5.(*Client).newRequest
|
||||
func newRequest(c *cos.Client, ctx context.Context, baseURL *url.URL, uri, method string, body interface{}, optQuery interface{}, optHeader interface{}) (req *http.Request, err error)
|
||||
@@ -1,15 +0,0 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package minio // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/minio"
|
||||
@@ -0,0 +1,11 @@
|
||||
package minio
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
_ "unsafe"
|
||||
|
||||
"github.com/minio/minio-go/v7"
|
||||
)
|
||||
|
||||
//go:linkname makeTargetURL github.com/minio/minio-go/v7.(*Client).makeTargetURL
|
||||
func makeTargetURL(client *minio.Client, bucketName, objectName, bucketLocation string, isVirtualHostStyle bool, queryValues url.Values) (*url.URL, error)
|
||||
@@ -139,6 +139,15 @@ func (m *Minio) initMinio(ctx context.Context) error {
|
||||
return fmt.Errorf("make bucket error: %w", err)
|
||||
}
|
||||
}
|
||||
if conf.PublicRead {
|
||||
policy := fmt.Sprintf(
|
||||
`{"Version": "2012-10-17","Statement": [{"Action": ["s3:GetObject","s3:PutObject"],"Effect": "Allow","Principal": {"AWS": ["*"]},"Resource": ["arn:aws:s3:::%s/*"],"Sid": ""}]}`,
|
||||
conf.Bucket,
|
||||
)
|
||||
if err := m.core.Client.SetBucketPolicy(ctx, conf.Bucket, policy); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
m.location, err = m.core.Client.GetBucketLocation(ctx, conf.Bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -375,7 +384,15 @@ func (m *Minio) presignedGetObject(ctx context.Context, name string, expire time
|
||||
} else if expire < time.Second {
|
||||
expire = time.Second
|
||||
}
|
||||
rawURL, err := m.sign.PresignedGetObject(ctx, m.bucket, name, expire, query)
|
||||
var (
|
||||
rawURL *url.URL
|
||||
err error
|
||||
)
|
||||
if config.Config.Object.Minio.PublicRead {
|
||||
rawURL, err = makeTargetURL(m.sign, m.bucket, name, m.location, false, query)
|
||||
} else {
|
||||
rawURL, err = m.sign.PresignedGetObject(ctx, m.bucket, name, expire, query)
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package oss // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/oss"
|
||||
@@ -16,10 +16,24 @@ package oss
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
_ "unsafe"
|
||||
|
||||
"github.com/aliyun/aliyun-oss-go-sdk/oss"
|
||||
)
|
||||
|
||||
//go:linkname ossSignHeader github.com/aliyun/aliyun-oss-go-sdk/oss.(*Conn).signHeader
|
||||
func ossSignHeader(c *oss.Conn, req *http.Request, canonicalizedResource string)
|
||||
//go:linkname signHeader github.com/aliyun/aliyun-oss-go-sdk/oss.Conn.signHeader
|
||||
func signHeader(c oss.Conn, req *http.Request, canonicalizedResource string)
|
||||
|
||||
//go:linkname getURLParams github.com/aliyun/aliyun-oss-go-sdk/oss.Conn.getURLParams
|
||||
func getURLParams(c oss.Conn, params map[string]interface{}) string
|
||||
|
||||
//go:linkname getURL github.com/aliyun/aliyun-oss-go-sdk/oss.urlMaker.getURL
|
||||
func getURL(um urlMaker, bucket, object, params string) *url.URL
|
||||
|
||||
type urlMaker struct {
|
||||
Scheme string
|
||||
NetLoc string
|
||||
Type int
|
||||
IsProxy bool
|
||||
}
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -69,6 +70,7 @@ func NewOSS() (s3.Interface, error) {
|
||||
bucketURL: conf.BucketURL,
|
||||
bucket: bucket,
|
||||
credentials: client.Config.GetCredentials(),
|
||||
um: *(*urlMaker)(reflect.ValueOf(bucket.Client.Conn).Elem().FieldByName("url").UnsafePointer()),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -76,6 +78,7 @@ type OSS struct {
|
||||
bucketURL string
|
||||
bucket *oss.Bucket
|
||||
credentials oss.Credentials
|
||||
um urlMaker
|
||||
}
|
||||
|
||||
func (o *OSS) Engine() string {
|
||||
@@ -163,7 +166,7 @@ func (o *OSS) AuthSign(ctx context.Context, uploadID string, name string, expire
|
||||
request.Header.Set(oss.HTTPHeaderHost, request.Host)
|
||||
request.Header.Set(oss.HTTPHeaderDate, now)
|
||||
request.Header.Set(oss.HttpHeaderOssDate, now)
|
||||
ossSignHeader(o.bucket.Client.Conn, request, fmt.Sprintf(`/%s/%s?partNumber=%d&uploadId=%s`, o.bucket.BucketName, name, partNumber, uploadID))
|
||||
signHeader(*o.bucket.Client.Conn, request, fmt.Sprintf(`/%s/%s?partNumber=%d&uploadId=%s`, o.bucket.BucketName, name, partNumber, uploadID))
|
||||
delete(request.Header, oss.HTTPHeaderDate)
|
||||
result.Parts[i] = s3.SignPart{
|
||||
PartNumber: partNumber,
|
||||
@@ -272,6 +275,7 @@ func (o *OSS) ListUploadedParts(ctx context.Context, uploadID string, name strin
|
||||
}
|
||||
|
||||
func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (string, error) {
|
||||
publicRead := config.Config.Object.Oss.PublicRead
|
||||
var opts []oss.Option
|
||||
if opt != nil {
|
||||
if opt.Image != nil {
|
||||
@@ -299,11 +303,13 @@ func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration,
|
||||
process += ",format," + format
|
||||
opts = append(opts, oss.Process(process))
|
||||
}
|
||||
if opt.ContentType != "" {
|
||||
opts = append(opts, oss.ResponseContentType(opt.ContentType))
|
||||
}
|
||||
if opt.Filename != "" {
|
||||
opts = append(opts, oss.ResponseContentDisposition(`attachment; filename=`+strconv.Quote(opt.Filename)))
|
||||
if !publicRead {
|
||||
if opt.ContentType != "" {
|
||||
opts = append(opts, oss.ResponseContentType(opt.ContentType))
|
||||
}
|
||||
if opt.Filename != "" {
|
||||
opts = append(opts, oss.ResponseContentDisposition(`attachment; filename=`+strconv.Quote(opt.Filename)))
|
||||
}
|
||||
}
|
||||
}
|
||||
if expire <= 0 {
|
||||
@@ -311,5 +317,13 @@ func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration,
|
||||
} else if expire < time.Second {
|
||||
expire = time.Second
|
||||
}
|
||||
return o.bucket.SignURL(name, http.MethodGet, int64(expire/time.Second), opts...)
|
||||
if !publicRead {
|
||||
return o.bucket.SignURL(name, http.MethodGet, int64(expire/time.Second), opts...)
|
||||
}
|
||||
rawParams, err := oss.GetRawParams(opts)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
params := getURLParams(*o.bucket.Client.Conn, rawParams)
|
||||
return getURL(o.um, o.bucket.BucketName, name, params).String(), nil
|
||||
}
|
||||
|
||||
@@ -66,5 +66,6 @@ type ConversationModelInterface interface {
|
||||
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (hashReadSeqs map[string]int64, err error)
|
||||
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*ConversationModel, error)
|
||||
GetConversationIDsNeedDestruct(ctx context.Context) ([]*ConversationModel, error)
|
||||
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
|
||||
NewTx(tx any) ConversationModelInterface
|
||||
}
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
package relation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Log struct {
|
||||
LogID string `gorm:"column:log_id;primary_key;type:char(64)"`
|
||||
Platform string `gorm:"column:platform;type:varchar(32)"`
|
||||
UserID string `gorm:"column:user_id;type:char(64)"`
|
||||
CreateTime time.Time `gorm:"index:,sort:desc"`
|
||||
Url string `gorm:"column:url;type varchar(255)"`
|
||||
FileName string `gorm:"column:filename;type varchar(255)"`
|
||||
SystemType string `gorm:"column:system_type;type varchar(255)"`
|
||||
Version string `gorm:"column:version;type varchar(255)"`
|
||||
Ex string `gorm:"column:ex;type varchar(255)"`
|
||||
}
|
||||
|
||||
type LogInterface interface {
|
||||
Create(ctx context.Context, log []*Log) error
|
||||
Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*Log, error)
|
||||
Delete(ctx context.Context, logID []string, userID string) error
|
||||
Get(ctx context.Context, logIDs []string, userID string) ([]*Log, error)
|
||||
}
|
||||
@@ -176,7 +176,7 @@ readonly API_URL=${API_URL:-"http://${IP}:${API_OPENIM_PORT}"}
|
||||
|
||||
def "OBJECT_ENABLE" "minio" # 对象是否启用
|
||||
# 对象的API地址
|
||||
readonly OBJECT_APIURL=${OBJECT_APIURL:-"http://${API_URL}"}
|
||||
readonly OBJECT_APIURL=${OBJECT_APIURL:-"${API_URL}"}
|
||||
def "MINIO_BUCKET" "openim" # MinIO的存储桶名称
|
||||
def "MINIO_PORT" "10005" # MinIO的端口
|
||||
# MinIO的端点URL
|
||||
@@ -186,17 +186,21 @@ def "MINIO_ACCESS_KEY" "${USER}"
|
||||
def "MINIO_SECRET_KEY" "${PASSWORD}" # MinIO的密钥
|
||||
def "MINIO_SESSION_TOKEN" # MinIO的会话令牌
|
||||
readonly MINIO_SIGN_ENDPOINT=${MINIO_SIGN_ENDPOINT:-"http://${IP}:${MINIO_PORT}"} # signEndpoint为minio公网地址
|
||||
def "MINIO_PUBLIC_READ" "false" # 公有读
|
||||
|
||||
# 腾讯云COS的存储桶URL
|
||||
def "COS_BUCKET_URL" "https://temp-1252357374.cos.ap-chengdu.myqcloud.com"
|
||||
def "COS_SECRET_ID" # 腾讯云COS的密钥ID
|
||||
def "COS_SECRET_KEY" # 腾讯云COS的密钥
|
||||
def "COS_SESSION_TOKEN" # 腾讯云COS的会话令牌
|
||||
def "COS_PUBLIC_READ" "false" # 公有读
|
||||
def "OSS_ENDPOINT" "https://oss-cn-chengdu.aliyuncs.com" # 阿里云OSS的端点URL
|
||||
def "OSS_BUCKET" "demo-9999999" # 阿里云OSS的存储桶名称
|
||||
def "OSS_BUCKET_URL" "https://demo-9999999.oss-cn-chengdu.aliyuncs.com" # 阿里云OSS的存储桶URL
|
||||
def "OSS_ACCESS_KEY_ID" # 阿里云OSS的访问密钥ID
|
||||
def "OSS_ACCESS_KEY_SECRET" # 阿里云OSS的密钥
|
||||
def "OSS_SESSION_TOKEN" # 阿里云OSS的会话令牌
|
||||
def "OSS_PUBLIC_READ" "false" # 公有读
|
||||
|
||||
###################### Redis 配置信息 ######################
|
||||
def "REDIS_PORT" "16379" # Redis的端口
|
||||
|
||||
@@ -22,7 +22,6 @@ import (
|
||||
"net"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio-go/v7"
|
||||
@@ -117,16 +116,24 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
func exactIP(urll string) string {
|
||||
u, _ := url.Parse(urll)
|
||||
host, _, err := net.SplitHostPort(u.Host)
|
||||
if err != nil {
|
||||
host = u.Host
|
||||
func checkMinioIP() error {
|
||||
for _, i := range []string{config.Config.Object.ApiURL, config.Config.Object.Minio.SignEndpoint} {
|
||||
u, err := url.Parse(i)
|
||||
if err != nil {
|
||||
return utils.Wrap(err, "api format error,please check config file apiURL or Minio SignEndpoint")
|
||||
}
|
||||
if u.Scheme == "https" {
|
||||
continue
|
||||
}
|
||||
host, _, err := net.SplitHostPort(u.Host)
|
||||
if err != nil {
|
||||
host = u.Host
|
||||
}
|
||||
if host == "127.0.0.1" {
|
||||
return ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1,please modify it")
|
||||
}
|
||||
}
|
||||
if strings.HasSuffix(host, ":") {
|
||||
host = host[0 : len(host)-1]
|
||||
}
|
||||
return host
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkMysql() error {
|
||||
@@ -205,8 +212,8 @@ func checkMinio() error {
|
||||
return ErrComponentStart.Wrap("Minio server is offline")
|
||||
}
|
||||
}
|
||||
if exactIP(config.Config.Object.ApiURL) == "127.0.0.1" || exactIP(config.Config.Object.Minio.SignEndpoint) == "127.0.0.1" {
|
||||
return ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1")
|
||||
if checkMinioIP() != nil {
|
||||
return checkMinioIP()
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
+3
-2
@@ -1,4 +1,4 @@
|
||||
module github.com/openimsdk/open-im-server/v3/tools/imctl
|
||||
module github.com/openimsdk/open-im-server/v3/tools/imctl
|
||||
|
||||
go 1.18
|
||||
|
||||
@@ -14,5 +14,6 @@ require (
|
||||
require (
|
||||
github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
|
||||
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
||||
golang.org/x/sys v0.1.0 // indirect
|
||||
golang.org/x/sys v0.10.0 // indirect
|
||||
k8s.io/kubernetes v1.28.2
|
||||
)
|
||||
|
||||
@@ -20,5 +20,8 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
|
||||
golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
|
||||
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
k8s.io/kubernetes v1.28.2 h1:GhcnYeNTukeaC0dD5BC+UWBvzQsFEpWj7XBVMQptfYc=
|
||||
k8s.io/kubernetes v1.28.2/go.mod h1:FmB1Mlp9ua0ezuwQCTGs/y6wj/fVisN2sVxhzjj0WDk=
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
module github.com/openimsdk/open-im-server/v3/tools/url2im
|
||||
|
||||
go 1.20
|
||||
|
||||
require (
|
||||
github.com/OpenIMSDK/protocol v0.0.21
|
||||
github.com/kelindar/bitmap v1.5.1
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.3 // indirect
|
||||
github.com/kelindar/simd v1.1.2 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.4 // indirect
|
||||
golang.org/x/net v0.9.0 // indirect
|
||||
golang.org/x/sys v0.7.0 // indirect
|
||||
golang.org/x/text v0.9.0 // indirect
|
||||
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
|
||||
google.golang.org/grpc v1.56.2 // indirect
|
||||
google.golang.org/protobuf v1.31.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,33 @@
|
||||
github.com/OpenIMSDK/protocol v0.0.21 h1:5H6H+hJ9d/VgRqttvxD/zfK9Asd+4M8Eknk5swSbUVY=
|
||||
github.com/OpenIMSDK/protocol v0.0.21/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
|
||||
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
|
||||
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
github.com/kelindar/bitmap v1.5.1 h1:+ZmZdwHbJ+CGE+q/aAJ74KJSnp0vOlGD7KY5x51mVzk=
|
||||
github.com/kelindar/bitmap v1.5.1/go.mod h1:j3qZjxH9s4OtvsnFTP2bmPkjqil9Y2xQlxPYHexasEA=
|
||||
github.com/kelindar/simd v1.1.2 h1:KduKb+M9cMY2HIH8S/cdJyD+5n5EGgq+Aeeleos55To=
|
||||
github.com/kelindar/simd v1.1.2/go.mod h1:inq4DFudC7W8L5fhxoeZflLRNpWSs0GNx6MlWFvuvr0=
|
||||
github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
|
||||
github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
|
||||
golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
|
||||
golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
|
||||
golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
|
||||
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
|
||||
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 h1:KpwkzHKEF7B9Zxg18WzOa7djJ+Ha5DzthMyZYQfEn2A=
|
||||
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1/go.mod h1:nKE/iIaLqn2bQwXBg8f1g2Ylh6r5MN5CmZvuzZCgsCU=
|
||||
google.golang.org/grpc v1.56.2 h1:fVRFRnXvU+x6C4IlHZewvJOVHoOv1TUuQyoRsYnB4bI=
|
||||
google.golang.org/grpc v1.56.2/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
|
||||
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
|
||||
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
|
||||
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
|
||||
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
@@ -0,0 +1,84 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/tools/url2im/pkg"
|
||||
)
|
||||
|
||||
/*take.txt
|
||||
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
|
||||
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
|
||||
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
|
||||
*/
|
||||
|
||||
func main() {
|
||||
var conf pkg.Config // 后面带*的为必填项
|
||||
flag.StringVar(&conf.TaskPath, "task", "take.txt", "task path") // 任务日志文件*
|
||||
flag.StringVar(&conf.ProgressPath, "progress", "", "progress path") // 进度日志文件
|
||||
flag.IntVar(&conf.Concurrency, "concurrency", 1, "concurrency num") // 并发数
|
||||
flag.IntVar(&conf.Retry, "retry", 1, "retry num") // 重试次数
|
||||
flag.StringVar(&conf.TempDir, "temp", "", "temp dir") // 临时文件夹
|
||||
flag.Int64Var(&conf.CacheSize, "cache", 1024*1024*100, "cache size") // 缓存大小(超过时,下载到磁盘)
|
||||
flag.Int64Var((*int64)(&conf.Timeout), "timeout", 5000, "timeout") // 请求超时时间(毫秒)
|
||||
flag.StringVar(&conf.Api, "api", "http://127.0.0.1:10002", "api") // im地址*
|
||||
flag.StringVar(&conf.UserID, "userID", "openIM123456", "userID") // im管理员
|
||||
flag.StringVar(&conf.Secret, "secret", "openIM123", "secret") // im config secret
|
||||
flag.Parse()
|
||||
if !filepath.IsAbs(conf.TaskPath) {
|
||||
var err error
|
||||
conf.TaskPath, err = filepath.Abs(conf.TaskPath)
|
||||
if err != nil {
|
||||
log.Println("get abs path err:", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
if conf.ProgressPath == "" {
|
||||
conf.ProgressPath = conf.TaskPath + ".progress.txt"
|
||||
} else if !filepath.IsAbs(conf.ProgressPath) {
|
||||
var err error
|
||||
conf.ProgressPath, err = filepath.Abs(conf.ProgressPath)
|
||||
if err != nil {
|
||||
log.Println("get abs path err:", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
if conf.TempDir == "" {
|
||||
conf.TempDir = conf.TaskPath + ".temp"
|
||||
}
|
||||
if info, err := os.Stat(conf.TempDir); err == nil {
|
||||
if !info.IsDir() {
|
||||
log.Printf("temp dir %s is not dir\n", err)
|
||||
return
|
||||
}
|
||||
} else if os.IsNotExist(err) {
|
||||
if err := os.MkdirAll(conf.TempDir, os.ModePerm); err != nil {
|
||||
log.Printf("mkdir temp dir %s err %+v\n", conf.TempDir, err)
|
||||
return
|
||||
}
|
||||
defer os.RemoveAll(conf.TempDir)
|
||||
} else {
|
||||
log.Println("get temp dir err:", err)
|
||||
return
|
||||
}
|
||||
if conf.Concurrency <= 0 {
|
||||
conf.Concurrency = 1
|
||||
}
|
||||
if conf.Retry <= 0 {
|
||||
conf.Retry = 1
|
||||
}
|
||||
if conf.CacheSize <= 0 {
|
||||
conf.CacheSize = 1024 * 1024 * 100 // 100M
|
||||
}
|
||||
if conf.Timeout <= 0 {
|
||||
conf.Timeout = 5000
|
||||
}
|
||||
conf.Timeout = conf.Timeout * time.Millisecond
|
||||
if err := pkg.Run(conf); err != nil {
|
||||
log.Println("main err:", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package pkg
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/auth"
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/OpenIMSDK/protocol/third"
|
||||
)
|
||||
|
||||
type Api struct {
|
||||
Api string
|
||||
UserID string
|
||||
Secret string
|
||||
Token string
|
||||
Client *http.Client
|
||||
}
|
||||
|
||||
func (a *Api) apiPost(ctx context.Context, path string, req any, resp any) error {
|
||||
operationID, _ := ctx.Value("operationID").(string)
|
||||
if operationID == "" {
|
||||
return errors.New("call api operationID is empty")
|
||||
}
|
||||
reqBody, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.Api+path, bytes.NewReader(reqBody))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
DefaultRequestHeader(request.Header)
|
||||
request.ContentLength = int64(len(reqBody))
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
request.Header.Set("operationID", operationID)
|
||||
if a.Token != "" {
|
||||
request.Header.Set("token", a.Token)
|
||||
}
|
||||
response, err := a.Client.Do(request)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
body, err := io.ReadAll(response.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("api %s status %s body %s", path, response.Status, body)
|
||||
}
|
||||
var baseResponse struct {
|
||||
ErrCode int `json:"errCode"`
|
||||
ErrMsg string `json:"errMsg"`
|
||||
ErrDlt string `json:"errDlt"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &baseResponse); err != nil {
|
||||
return err
|
||||
}
|
||||
if baseResponse.ErrCode != 0 {
|
||||
return fmt.Errorf("api %s errCode %d errMsg %s errDlt %s", path, baseResponse.ErrCode, baseResponse.ErrMsg, baseResponse.ErrDlt)
|
||||
}
|
||||
if resp != nil {
|
||||
if err := json.Unmarshal(baseResponse.Data, resp); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Api) GetToken(ctx context.Context) (string, error) {
|
||||
req := auth.UserTokenReq{
|
||||
UserID: a.UserID,
|
||||
Secret: a.Secret,
|
||||
PlatformID: constant.AdminPlatformID,
|
||||
}
|
||||
var resp auth.UserTokenResp
|
||||
if err := a.apiPost(ctx, "/auth/user_token", &req, &resp); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resp.Token, nil
|
||||
}
|
||||
|
||||
func (a *Api) GetPartLimit(ctx context.Context) (*third.PartLimitResp, error) {
|
||||
var resp third.PartLimitResp
|
||||
if err := a.apiPost(ctx, "/object/part_limit", &third.PartLimitReq{}, &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
func (a *Api) InitiateMultipartUpload(ctx context.Context, req *third.InitiateMultipartUploadReq) (*third.InitiateMultipartUploadResp, error) {
|
||||
var resp third.InitiateMultipartUploadResp
|
||||
if err := a.apiPost(ctx, "/object/initiate_multipart_upload", req, &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
func (a *Api) CompleteMultipartUpload(ctx context.Context, req *third.CompleteMultipartUploadReq) (string, error) {
|
||||
var resp third.CompleteMultipartUploadResp
|
||||
if err := a.apiPost(ctx, "/object/complete_multipart_upload", req, &resp); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resp.Url, nil
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
package pkg
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"os"
|
||||
)
|
||||
|
||||
type ReadSeekSizeCloser interface {
|
||||
io.ReadSeekCloser
|
||||
Size() int64
|
||||
}
|
||||
|
||||
func NewReader(r io.Reader, max int64, path string) (ReadSeekSizeCloser, error) {
|
||||
buf := make([]byte, max+1)
|
||||
n, err := io.ReadFull(r, buf)
|
||||
if err == nil {
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0666)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var ok bool
|
||||
defer func() {
|
||||
if !ok {
|
||||
_ = f.Close()
|
||||
_ = os.Remove(path)
|
||||
}
|
||||
}()
|
||||
if _, err := f.Write(buf[:n]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cn, err := io.Copy(f, r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, err := f.Seek(0, io.SeekStart); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ok = true
|
||||
return &fileBuffer{
|
||||
f: f,
|
||||
n: cn + int64(n),
|
||||
}, nil
|
||||
} else if err == io.EOF || err == io.ErrUnexpectedEOF {
|
||||
return &memoryBuffer{
|
||||
r: bytes.NewReader(buf[:n]),
|
||||
}, nil
|
||||
} else {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
type fileBuffer struct {
|
||||
n int64
|
||||
f *os.File
|
||||
}
|
||||
|
||||
func (r *fileBuffer) Read(p []byte) (n int, err error) {
|
||||
return r.f.Read(p)
|
||||
}
|
||||
|
||||
func (r *fileBuffer) Seek(offset int64, whence int) (int64, error) {
|
||||
return r.f.Seek(offset, whence)
|
||||
}
|
||||
|
||||
func (r *fileBuffer) Size() int64 {
|
||||
return r.n
|
||||
}
|
||||
|
||||
func (r *fileBuffer) Close() error {
|
||||
name := r.f.Name()
|
||||
if err := r.f.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Remove(name)
|
||||
}
|
||||
|
||||
type memoryBuffer struct {
|
||||
r *bytes.Reader
|
||||
}
|
||||
|
||||
func (r *memoryBuffer) Read(p []byte) (n int, err error) {
|
||||
return r.r.Read(p)
|
||||
}
|
||||
|
||||
func (r *memoryBuffer) Seek(offset int64, whence int) (int64, error) {
|
||||
return r.r.Seek(offset, whence)
|
||||
}
|
||||
|
||||
func (r *memoryBuffer) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *memoryBuffer) Size() int64 {
|
||||
return r.r.Size()
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package pkg
|
||||
|
||||
import "time"
|
||||
|
||||
type Config struct {
|
||||
TaskPath string
|
||||
ProgressPath string
|
||||
Concurrency int
|
||||
Retry int
|
||||
Timeout time.Duration
|
||||
Api string
|
||||
UserID string
|
||||
Secret string
|
||||
TempDir string
|
||||
CacheSize int64
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package pkg
|
||||
|
||||
import "net/http"
|
||||
|
||||
func DefaultRequestHeader(header http.Header) {
|
||||
header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36")
|
||||
}
|
||||
@@ -0,0 +1,385 @@
|
||||
package pkg
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/third"
|
||||
)
|
||||
|
||||
type Upload struct {
|
||||
URL string `json:"url"`
|
||||
Name string `json:"name"`
|
||||
ContentType string `json:"contentType"`
|
||||
}
|
||||
|
||||
type Task struct {
|
||||
Index int
|
||||
Upload Upload
|
||||
}
|
||||
|
||||
type PartInfo struct {
|
||||
ContentType string
|
||||
PartSize int64
|
||||
PartNum int
|
||||
FileMd5 string
|
||||
PartMd5 string
|
||||
PartSizes []int64
|
||||
PartMd5s []string
|
||||
}
|
||||
|
||||
func Run(conf Config) error {
|
||||
m := &Manage{
|
||||
prefix: time.Now().Format("20060102150405"),
|
||||
conf: &conf,
|
||||
ctx: context.Background(),
|
||||
}
|
||||
return m.Run()
|
||||
}
|
||||
|
||||
type Manage struct {
|
||||
conf *Config
|
||||
ctx context.Context
|
||||
api *Api
|
||||
partLimit *third.PartLimitResp
|
||||
prefix string
|
||||
tasks chan Task
|
||||
id uint64
|
||||
success int64
|
||||
failed int64
|
||||
}
|
||||
|
||||
func (m *Manage) tempFilePath() string {
|
||||
return filepath.Join(m.conf.TempDir, fmt.Sprintf("%s_%d", m.prefix, atomic.AddUint64(&m.id, 1)))
|
||||
}
|
||||
|
||||
func (m *Manage) Run() error {
|
||||
defer func(start time.Time) {
|
||||
log.Printf("run time %s\n", time.Since(start))
|
||||
}(time.Now())
|
||||
m.api = &Api{
|
||||
Api: m.conf.Api,
|
||||
UserID: m.conf.UserID,
|
||||
Secret: m.conf.Secret,
|
||||
Client: &http.Client{Timeout: m.conf.Timeout},
|
||||
}
|
||||
var err error
|
||||
ctx := context.WithValue(m.ctx, "operationID", fmt.Sprintf("%s_init", m.prefix))
|
||||
m.api.Token, err = m.api.GetToken(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.partLimit, err = m.api.GetPartLimit(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
progress, err := ReadProgress(m.conf.ProgressPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
progressFile, err := os.OpenFile(m.conf.ProgressPath, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var mutex sync.Mutex
|
||||
writeSuccessIndex := func(index int) {
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
if _, err := progressFile.Write([]byte(strconv.Itoa(index) + "\n")); err != nil {
|
||||
log.Printf("write progress err: %v\n", err)
|
||||
}
|
||||
}
|
||||
file, err := os.Open(m.conf.TaskPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.tasks = make(chan Task, m.conf.Concurrency*2)
|
||||
go func() {
|
||||
defer file.Close()
|
||||
defer close(m.tasks)
|
||||
scanner := bufio.NewScanner(file)
|
||||
var (
|
||||
index int
|
||||
num int
|
||||
)
|
||||
for scanner.Scan() {
|
||||
line := strings.TrimSpace(scanner.Text())
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
index++
|
||||
if progress.IsUploaded(index) {
|
||||
log.Printf("index: %d already uploaded %s\n", index, line)
|
||||
continue
|
||||
}
|
||||
var upload Upload
|
||||
if err := json.Unmarshal([]byte(line), &upload); err != nil {
|
||||
log.Printf("index: %d json.Unmarshal(%s) err: %v", index, line, err)
|
||||
continue
|
||||
}
|
||||
num++
|
||||
m.tasks <- Task{
|
||||
Index: index,
|
||||
Upload: upload,
|
||||
}
|
||||
}
|
||||
if num == 0 {
|
||||
log.Println("mark all completed")
|
||||
}
|
||||
}()
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(m.conf.Concurrency)
|
||||
for i := 0; i < m.conf.Concurrency; i++ {
|
||||
go func(tid int) {
|
||||
defer wg.Done()
|
||||
for task := range m.tasks {
|
||||
var success bool
|
||||
for n := 0; n < m.conf.Retry; n++ {
|
||||
ctx := context.WithValue(m.ctx, "operationID", fmt.Sprintf("%s_%d_%d_%d", m.prefix, tid, task.Index, n+1))
|
||||
if urlRaw, err := m.RunTask(ctx, task); err == nil {
|
||||
writeSuccessIndex(task.Index)
|
||||
log.Println("index:", task.Index, "upload success", "urlRaw", urlRaw)
|
||||
success = true
|
||||
break
|
||||
} else {
|
||||
log.Printf("index: %d upload: %+v err: %v", task.Index, task.Upload, err)
|
||||
}
|
||||
}
|
||||
if success {
|
||||
atomic.AddInt64(&m.success, 1)
|
||||
} else {
|
||||
atomic.AddInt64(&m.failed, 1)
|
||||
log.Printf("index: %d upload: %+v failed", task.Index, task.Upload)
|
||||
}
|
||||
}
|
||||
}(i + 1)
|
||||
}
|
||||
wg.Wait()
|
||||
log.Printf("execution completed success %d failed %d\n", m.success, m.failed)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manage) RunTask(ctx context.Context, task Task) (string, error) {
|
||||
resp, err := m.HttpGet(ctx, task.Upload.URL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
reader, err := NewReader(resp.Body, m.conf.CacheSize, m.tempFilePath())
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer reader.Close()
|
||||
part, err := m.getPartInfo(ctx, reader, reader.Size())
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var contentType string
|
||||
if task.Upload.ContentType == "" {
|
||||
contentType = part.ContentType
|
||||
} else {
|
||||
contentType = task.Upload.ContentType
|
||||
}
|
||||
initiateMultipartUploadResp, err := m.api.InitiateMultipartUpload(ctx, &third.InitiateMultipartUploadReq{
|
||||
Hash: part.PartMd5,
|
||||
Size: reader.Size(),
|
||||
PartSize: part.PartSize,
|
||||
MaxParts: -1,
|
||||
Cause: "batch-import",
|
||||
Name: task.Upload.Name,
|
||||
ContentType: contentType,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if initiateMultipartUploadResp.Upload == nil {
|
||||
return initiateMultipartUploadResp.Url, nil
|
||||
}
|
||||
if _, err := reader.Seek(0, io.SeekStart); err != nil {
|
||||
return "", err
|
||||
}
|
||||
uploadParts := make([]*third.SignPart, part.PartNum)
|
||||
for _, part := range initiateMultipartUploadResp.Upload.Sign.Parts {
|
||||
uploadParts[part.PartNumber-1] = part
|
||||
}
|
||||
for i, currentPartSize := range part.PartSizes {
|
||||
md5Reader := NewMd5Reader(io.LimitReader(reader, currentPartSize))
|
||||
if m.doPut(ctx, m.api.Client, initiateMultipartUploadResp.Upload.Sign, uploadParts[i], md5Reader, currentPartSize); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if md5val := md5Reader.Md5(); md5val != part.PartMd5s[i] {
|
||||
return "", fmt.Errorf("upload part %d failed, md5 not match, expect %s, got %s", i, part.PartMd5s[i], md5val)
|
||||
}
|
||||
}
|
||||
urlRaw, err := m.api.CompleteMultipartUpload(ctx, &third.CompleteMultipartUploadReq{
|
||||
UploadID: initiateMultipartUploadResp.Upload.UploadID,
|
||||
Parts: part.PartMd5s,
|
||||
Name: task.Upload.Name,
|
||||
ContentType: contentType,
|
||||
Cause: "batch-import",
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return urlRaw, nil
|
||||
}
|
||||
|
||||
func (m *Manage) partSize(size int64) (int64, error) {
|
||||
if size <= 0 {
|
||||
return 0, errors.New("size must be greater than 0")
|
||||
}
|
||||
if size > m.partLimit.MaxPartSize*int64(m.partLimit.MaxNumSize) {
|
||||
return 0, fmt.Errorf("size must be less than %db", m.partLimit.MaxPartSize*int64(m.partLimit.MaxNumSize))
|
||||
}
|
||||
if size <= m.partLimit.MinPartSize*int64(m.partLimit.MaxNumSize) {
|
||||
return m.partLimit.MinPartSize, nil
|
||||
}
|
||||
partSize := size / int64(m.partLimit.MaxNumSize)
|
||||
if size%int64(m.partLimit.MaxNumSize) != 0 {
|
||||
partSize++
|
||||
}
|
||||
return partSize, nil
|
||||
}
|
||||
|
||||
func (m *Manage) partMD5(parts []string) string {
|
||||
s := strings.Join(parts, ",")
|
||||
md5Sum := md5.Sum([]byte(s))
|
||||
return hex.EncodeToString(md5Sum[:])
|
||||
}
|
||||
|
||||
func (m *Manage) getPartInfo(ctx context.Context, r io.Reader, fileSize int64) (*PartInfo, error) {
|
||||
partSize, err := m.partSize(fileSize)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
partNum := int(fileSize / partSize)
|
||||
if fileSize%partSize != 0 {
|
||||
partNum++
|
||||
}
|
||||
partSizes := make([]int64, partNum)
|
||||
for i := 0; i < partNum; i++ {
|
||||
partSizes[i] = partSize
|
||||
}
|
||||
partSizes[partNum-1] = fileSize - partSize*(int64(partNum)-1)
|
||||
partMd5s := make([]string, partNum)
|
||||
buf := make([]byte, 1024*8)
|
||||
fileMd5 := md5.New()
|
||||
var contentType string
|
||||
for i := 0; i < partNum; i++ {
|
||||
h := md5.New()
|
||||
r := io.LimitReader(r, partSize)
|
||||
for {
|
||||
if n, err := r.Read(buf); err == nil {
|
||||
if contentType == "" {
|
||||
contentType = http.DetectContentType(buf[:n])
|
||||
}
|
||||
h.Write(buf[:n])
|
||||
fileMd5.Write(buf[:n])
|
||||
} else if err == io.EOF {
|
||||
break
|
||||
} else {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
partMd5s[i] = hex.EncodeToString(h.Sum(nil))
|
||||
}
|
||||
partMd5Val := m.partMD5(partMd5s)
|
||||
fileMd5val := hex.EncodeToString(fileMd5.Sum(nil))
|
||||
return &PartInfo{
|
||||
ContentType: contentType,
|
||||
PartSize: partSize,
|
||||
PartNum: partNum,
|
||||
FileMd5: fileMd5val,
|
||||
PartMd5: partMd5Val,
|
||||
PartSizes: partSizes,
|
||||
PartMd5s: partMd5s,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *Manage) doPut(ctx context.Context, client *http.Client, sign *third.AuthSignParts, part *third.SignPart, reader io.Reader, size int64) error {
|
||||
rawURL := part.Url
|
||||
if rawURL == "" {
|
||||
rawURL = sign.Url
|
||||
}
|
||||
if len(sign.Query)+len(part.Query) > 0 {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
query := u.Query()
|
||||
for i := range sign.Query {
|
||||
v := sign.Query[i]
|
||||
query[v.Key] = v.Values
|
||||
}
|
||||
for i := range part.Query {
|
||||
v := part.Query[i]
|
||||
query[v.Key] = v.Values
|
||||
}
|
||||
u.RawQuery = query.Encode()
|
||||
rawURL = u.String()
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPut, rawURL, reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range sign.Header {
|
||||
v := sign.Header[i]
|
||||
req.Header[v.Key] = v.Values
|
||||
}
|
||||
for i := range part.Header {
|
||||
v := part.Header[i]
|
||||
req.Header[v.Key] = v.Values
|
||||
}
|
||||
req.ContentLength = size
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
}()
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.StatusCode/200 != 1 {
|
||||
return fmt.Errorf("PUT %s part %d failed, status code %d, body %s", rawURL, part.PartNumber, resp.StatusCode, string(body))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manage) HttpGet(ctx context.Context, url string) (*http.Response, error) {
|
||||
reqUrl := url
|
||||
for {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, reqUrl, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
DefaultRequestHeader(request.Header)
|
||||
response, err := m.api.Client.Do(request)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if response.StatusCode != http.StatusOK {
|
||||
_ = response.Body.Close()
|
||||
return nil, fmt.Errorf("http get %s status %s", url, response.Status)
|
||||
}
|
||||
return response, nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package pkg
|
||||
|
||||
import (
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"hash"
|
||||
"io"
|
||||
)
|
||||
|
||||
func NewMd5Reader(r io.Reader) *Md5Reader {
|
||||
return &Md5Reader{h: md5.New(), r: r}
|
||||
}
|
||||
|
||||
type Md5Reader struct {
|
||||
h hash.Hash
|
||||
r io.Reader
|
||||
}
|
||||
|
||||
func (r *Md5Reader) Read(p []byte) (n int, err error) {
|
||||
n, err = r.r.Read(p)
|
||||
if err == nil && n > 0 {
|
||||
r.h.Write(p[:n])
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (r *Md5Reader) Md5() string {
|
||||
return hex.EncodeToString(r.h.Sum(nil))
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package pkg
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"os"
|
||||
"strconv"
|
||||
|
||||
"github.com/kelindar/bitmap"
|
||||
)
|
||||
|
||||
func ReadProgress(path string) (*Progress, error) {
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return &Progress{}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
defer file.Close()
|
||||
scanner := bufio.NewScanner(file)
|
||||
var upload bitmap.Bitmap
|
||||
for scanner.Scan() {
|
||||
index, err := strconv.Atoi(scanner.Text())
|
||||
if err != nil || index < 0 {
|
||||
continue
|
||||
}
|
||||
upload.Set(uint32(index))
|
||||
}
|
||||
return &Progress{upload: upload}, nil
|
||||
}
|
||||
|
||||
type Progress struct {
|
||||
upload bitmap.Bitmap
|
||||
}
|
||||
|
||||
func (p *Progress) IsUploaded(index int) bool {
|
||||
if p == nil {
|
||||
return false
|
||||
}
|
||||
return p.upload.Contains(uint32(index))
|
||||
}
|
||||
Reference in New Issue
Block a user