IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-06 10:05:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

msg

Browse Source
This commit is contained in:
withchao
2023-02-23 11:26:46 +08:00
parent b4443c843a
commit eac2619c33
6 changed files with 148 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/rpc/msg/extend_msg.go
+36 -93
View File
@@ -14,107 +14,50 @@ import (
)
func (m *msgServer) SetMessageReactionExtensions(ctx context.Context, req *msg.SetMessageReactionExtensionsReq) (resp *msg.SetMessageReactionExtensionsResp, err error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "m args is:", req.String())
var rResp msg.SetMessageReactionExtensionsResp
rResp.ClientMsgID = req.ClientMsgID
rResp.MsgFirstModifyTime = req.MsgFirstModifyTime
callbackResp := notification.callbackSetMessageReactionExtensions(req)
if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 {
rResp.ErrCode = int32(callbackResp.ErrCode)
rResp.ErrMsg = callbackResp.ErrMsg
for _, value := range req.ReactionExtensionList {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = callbackResp.ErrMsg
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
return &rResp, nil
resp = &msg.SetMessageReactionExtensionsResp{}
//resp.ClientMsgID = req.ClientMsgID
//resp.MsgFirstModifyTime = req.MsgFirstModifyTime
if err := CallbackSetMessageReactionExtensions(ctx, req); err != nil {
return nil, err
}
//if ExternalExtension
if req.IsExternalExtensions {
var isHistory bool
if req.IsReact {
isHistory = false
} else {
isHistory = true
}
rResp.MsgFirstModifyTime = callbackResp.MsgFirstModifyTime
rResp.Result = callbackResp.ResultReactionExtensionList
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, isHistory, false)
return &rResp, nil
resp.MsgFirstModifyTime = req.MsgFirstModifyTime
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &resp, !req.IsReact, false)
return resp, nil
}
for _, v := range callbackResp.ResultReactionExtensionList {
if v.ErrCode == 0 {
req.ReactionExtensionList[v.KeyValue.TypeKey] = v.KeyValue
} else {
delete(req.ReactionExtensionList, v.KeyValue.TypeKey)
rResp.Result = append(rResp.Result, v)
}
}
isExists, err := db.DB.JudgeMessageReactionEXISTS(req.ClientMsgID, req.SessionType)
isExists, err := m.MsgInterface.JudgeMessageReactionEXISTS(ctx, req.ClientMsgID, req.SessionType)
if err != nil {
rResp.ErrCode = 100
rResp.ErrMsg = err.Error()
for _, value := range req.ReactionExtensionList {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
return &rResp, nil
return nil, err
}
if !isExists {
if !req.IsReact {
log.Debug(req.OperationID, "redis handle firstly", req.String())
rResp.MsgFirstModifyTime = utils.GetCurrentTimestampByMill()
resp.MsgFirstModifyTime = utils.GetCurrentTimestampByMill()
for k, v := range req.ReactionExtensionList {
err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
err := m.MessageLocker.LockMessageTypeKey(ctx, req.ClientMsgID, k)
if err != nil {
setKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, k, v)
continue
return nil, err
}
v.LatestUpdateTime = utils.GetCurrentTimestampByMill()
newerr := db.DB.SetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k, utils.StructToJsonString(v))
if newerr != nil {
setKeyResultInfo(&rResp, 201, newerr.Error(), req.ClientMsgID, k, v)
continue
if err := m.MsgInterface.SetMessageTypeKeyValue(ctx, req.ClientMsgID, req.SessionType, k, utils.StructToJsonString(v)); err != nil {
return nil, err
}
setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
}
rResp.IsReact = true
_, err := db.DB.SetMessageReactionExpire(req.ClientMsgID, req.SessionType, time.Duration(24*3)*time.Hour)
resp.IsReact = true
_, err := m.MsgInterface.SetMessageReactionExpire(ctx, req.ClientMsgID, req.SessionType, time.Duration(24*3)*time.Hour)
if err != nil {
log.Error(req.OperationID, "SetMessageReactionExpire err:", err.Error(), req.String())
return nil, err
}
} else {
err := m.dMessageLocker.LockGlobalMessage(req.ClientMsgID)
err := m.MessageLocker.LockGlobalMessage(ctx, req.ClientMsgID)
if err != nil {
rResp.ErrCode = 100
rResp.ErrMsg = err.Error()
for _, value := range req.ReactionExtensionList {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
return &rResp, nil
return nil, err
}
mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime)
mongoValue, err := m.MsgInterface.GetExtendMsg(ctx, req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime)
if err != nil {
rResp.ErrCode = 200
rResp.ErrMsg = err.Error()
for _, value := range req.ReactionExtensionList {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
return &rResp, nil
return nil, err
}
setValue := make(map[string]*sdkws.KeyValue)
for k, v := range req.ReactionExtensionList {
@@ -123,7 +66,7 @@ func (m *msgServer) SetMessageReactionExtensions(ctx context.Context, req *msg.S
if vv, ok := mongoValue.ReactionExtensionList[k]; ok {
utils.CopyStructFields(temp, &vv)
if v.LatestUpdateTime != vv.LatestUpdateTime {
setKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, k, temp)
setKeyResultInfo(&resp, 300, "message have update", req.ClientMsgID, k, temp)
continue
}
}
@@ -139,13 +82,13 @@ func (m *msgServer) SetMessageReactionExtensions(ctx context.Context, req *msg.S
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
resp.Result = append(resp.Result, temp)
}
} else {
for _, value := range setValue {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
rResp.Result = append(rResp.Result, temp)
resp.Result = append(resp.Result, temp)
}
}
lockErr := m.dMessageLocker.UnLockGlobalMessage(req.ClientMsgID)
@@ -160,42 +103,42 @@ func (m *msgServer) SetMessageReactionExtensions(ctx context.Context, req *msg.S
for k, v := range req.ReactionExtensionList {
err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
if err != nil {
setKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, k, v)
setKeyResultInfo(&resp, 100, err.Error(), req.ClientMsgID, k, v)
continue
}
redisValue, err := db.DB.GetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k)
if err != nil && err != go_redis.Nil {
setKeyResultInfo(&rResp, 200, err.Error(), req.ClientMsgID, k, v)
setKeyResultInfo(&resp, 200, err.Error(), req.ClientMsgID, k, v)
continue
}
temp := new(sdkws.KeyValue)
utils.JsonStringToStruct(redisValue, temp)
if v.LatestUpdateTime != temp.LatestUpdateTime {
setKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, k, temp)
setKeyResultInfo(&resp, 300, "message have update", req.ClientMsgID, k, temp)
continue
} else {
v.LatestUpdateTime = utils.GetCurrentTimestampByMill()
newerr := db.DB.SetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k, utils.StructToJsonString(v))
if newerr != nil {
setKeyResultInfo(&rResp, 201, newerr.Error(), req.ClientMsgID, k, temp)
setKeyResultInfo(&resp, 201, newerr.Error(), req.ClientMsgID, k, temp)
continue
}
setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
setKeyResultInfo(&resp, 0, "", req.ClientMsgID, k, v)
}
}
}
if !isExists {
if !req.IsReact {
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, true, true)
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &resp, true, true)
} else {
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, false)
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &resp, false, false)
}
} else {
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, true)
notification.ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &resp, false, true)
}
log.Debug(req.OperationID, utils.GetSelfFuncName(), "m return is:", rResp.String())
return &rResp, nil
log.Debug(req.OperationID, utils.GetSelfFuncName(), "m return is:", resp.String())
return &resp, nil
}
func setKeyResultInfo(r *msg.SetMessageReactionExtensionsResp, errCode int32, errMsg, clientMsgID, typeKey string, keyValue *sdkws.KeyValue) {
internal/rpc/msg/extend_msg_callback.go
+5 -1
View File
@@ -27,7 +27,11 @@ func CallbackSetMessageReactionExtensions(ctx context.Context, setReq *msg.SetMe
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbapi.CallbackBeforeSetMessageReactionExtResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
if err := http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
return err
}
setReq.MsgFirstModifyTime = resp.MsgFirstModifyTime
return nil
}
func CallbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReactionExtensionsReq) error {
internal/rpc/msg/lock.go
+5 -4
View File
@@ -1,16 +1,17 @@
package msg
import (
"context"
"time"
)
const GlOBLLOCK = "GLOBAL_LOCK"
type MessageLocker interface {
LockMessageTypeKey(clientMsgID, typeKey string) (err error)
UnLockMessageTypeKey(clientMsgID string, typeKey string) error
LockGlobalMessage(clientMsgID string) (err error)
UnLockGlobalMessage(clientMsgID string) (err error)
LockMessageTypeKey(ctx context.Context, clientMsgID, typeKey string) (err error)
UnLockMessageTypeKey(ctx context.Context, clientMsgID string, typeKey string) error
LockGlobalMessage(ctx context.Context, clientMsgID string) (err error)
UnLockGlobalMessage(ctx context.Context, clientMsgID string) (err error)
}
type LockerMessage struct{}
internal/rpc/msg/send_pull.go
+1
View File
@@ -13,6 +13,7 @@ import (
)
func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
resp = &msg.SendMsgResp{}
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
// callback
if err = CallbackBeforeSendGroupMsg(ctx, req); err != nil {
internal/rpc/msg/server.go
+3 -2
View File
@@ -16,13 +16,14 @@ import (
type msgServer struct {
RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
MsgInterface controller.MsgInterface
MsgInterface controller.MsgDatabaseInterface
Group *check.GroupChecker
User *check.UserCheck
Conversation *check.ConversationChecker
friend *check.FriendChecker
*localcache.GroupLocalCache
black *check.BlackChecker
black *check.BlackChecker
MessageLocker MessageLocker
}
type deleteMsg struct {