IM/open-im-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/v2.3.0release' into v2.3.0release

Browse Source
This commit is contained in:
Gordon
2023-02-01 17:37:47 +08:00
parent a60b584dca 3cb3b6ba2d
commit 396a87ab71
84 changed files with 8085 additions and 5713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/manage/management_chat.go
+3 -1
View File
@@ -38,6 +38,7 @@ func SetOptions(options map[string]bool, value bool) {
func newUserSendMsgReq(params *api.ManagementSendMsgReq) *pbChat.SendMsgReq {
var newContent string
options := make(map[string]bool, 5)
var err error
switch params.ContentType {
case constant.Text:
@@ -57,12 +58,13 @@ func newUserSendMsgReq(params *api.ManagementSendMsgReq) *pbChat.SendMsgReq {
case constant.CustomOnlineOnly:
fallthrough
case constant.AdvancedRevoke:
utils.SetSwitchFromOptions(options, constant.IsUnreadCount, false)
newContent = utils.StructToJsonString(params.Content)
case constant.Revoke:
utils.SetSwitchFromOptions(options, constant.IsUnreadCount, false)
newContent = params.Content["revokeMsgClientID"].(string)
default:
}
options := make(map[string]bool, 5)
if params.IsOnlineOnly {
SetOptions(options, false)
}
internal/api/msg/extend_msg.go
+207
View File
@@ -0,0 +1,207 @@
package msg
import (
api "Open_IM/pkg/base_info"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/token_verify"
"Open_IM/pkg/grpc-etcdv3/getcdv3"
rpc "Open_IM/pkg/proto/msg"
"Open_IM/pkg/utils"
"context"
"github.com/gin-gonic/gin"
"net/http"
"strings"
)
func SetMessageReactionExtensions(c *gin.Context) {
var (
req api.SetMessageReactionExtensionsReq
resp api.SetMessageReactionExtensionsResp
reqPb rpc.SetMessageReactionExtensionsReq
)
if err := c.BindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
if err := utils.CopyStructFields(&reqPb, &req); err != nil {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
}
var ok bool
var errInfo string
ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
if grpcConn == nil {
errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
msgClient := rpc.NewMsgClient(grpcConn)
respPb, err := msgClient.SetMessageReactionExtensions(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
return
}
resp.ErrCode = respPb.ErrCode
resp.ErrMsg = respPb.ErrMsg
resp.Data.ResultKeyValue = respPb.Result
resp.Data.MsgFirstModifyTime = reqPb.MsgFirstModifyTime
resp.Data.IsReact = reqPb.IsReact
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
c.JSON(http.StatusOK, resp)
}
func GetMessageListReactionExtensions(c *gin.Context) {
var (
req api.GetMessageListReactionExtensionsReq
resp api.GetMessageListReactionExtensionsResp
reqPb rpc.GetMessageListReactionExtensionsReq
)
if err := c.BindJSON(&req); err != nil {
c.JSON(http.StatusOK, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
if err := utils.CopyStructFields(&reqPb, &req); err != nil {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
}
var ok bool
var errInfo string
ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusOK, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
if grpcConn == nil {
errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
msgClient := rpc.NewMsgClient(grpcConn)
respPb, err := msgClient.GetMessageListReactionExtensions(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
return
}
resp.ErrCode = respPb.ErrCode
resp.ErrMsg = respPb.ErrMsg
resp.Data = respPb.SingleMessageResult
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
c.JSON(http.StatusOK, resp)
}
func AddMessageReactionExtensions(c *gin.Context) {
var (
req api.AddMessageReactionExtensionsReq
resp api.AddMessageReactionExtensionsResp
reqPb rpc.AddMessageReactionExtensionsReq
)
if err := c.BindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
if err := utils.CopyStructFields(&reqPb, &req); err != nil {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
}
var ok bool
var errInfo string
ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
if grpcConn == nil {
errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
msgClient := rpc.NewMsgClient(grpcConn)
respPb, err := msgClient.AddMessageReactionExtensions(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
return
}
resp.ErrCode = respPb.ErrCode
resp.ErrMsg = respPb.ErrMsg
resp.Data.ResultKeyValue = respPb.Result
resp.Data.MsgFirstModifyTime = respPb.MsgFirstModifyTime
resp.Data.IsReact = respPb.IsReact
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
c.JSON(http.StatusOK, resp)
}
func DeleteMessageReactionExtensions(c *gin.Context) {
var (
req api.DeleteMessageReactionExtensionsReq
resp api.DeleteMessageReactionExtensionsResp
reqPb rpc.DeleteMessageListReactionExtensionsReq
)
if err := c.BindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
if err := utils.CopyStructFields(&reqPb, &req); err != nil {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
}
var ok bool
var errInfo string
ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
if grpcConn == nil {
errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
msgClient := rpc.NewMsgClient(grpcConn)
respPb, err := msgClient.DeleteMessageReactionExtensions(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
return
}
resp.ErrCode = respPb.ErrCode
resp.ErrMsg = respPb.ErrMsg
resp.Data = respPb.Result
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
c.JSON(http.StatusOK, resp)
}
internal/cron_task/clear_msg.go
+38 -59
View File
@@ -7,11 +7,12 @@ import (
"Open_IM/pkg/common/log"
server_api_params "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
goRedis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
"math"
"strconv"
"strings"
goRedis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
)
const oldestList = 0
@@ -105,59 +106,45 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
if len(msgs.Msg) > db.GetSingleGocMsgNum() {
log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID)
}
var hasMsgDoNotNeedDel bool
for i, msg := range msgs.Msg {
// 找到列表中不需要删除的消息了, 表示为递归到最后一个块
if utils.GetCurrentTimestampByMill() < msg.SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) {
log.NewDebug(operationID, ID, "find uid", msgs.UID)
// 删除块失败 递归结束 返回0
hasMsgDoNotNeedDel = true
if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil {
return 0, err
}
// unMarshall失败 块删除成功 设置为最小seq
msgPb := &server_api_params.MsgData{}
if err = proto.Unmarshal(msg.Msg, msgPb); err != nil {
return delStruct.getSetMinSeq(), utils.Wrap(err, "")
}
// 如果不是块中第一个，就把前面比他早插入的全部设置空 seq字段除外。
if i > 0 {
delStruct.minSeq, err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i)
return delStruct.getSetMinSeq(), utils.Wrap(err, "")
}
}
// 递归结束
return msgPb.Seq, nil
} else {
if !msgListIsFull(msgs) {
}
}
}
// 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归
lastMsgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
delStruct.minSeq = lastMsgPb.Seq
if msgListIsFull(msgs) {
log.NewDebug(operationID, "msg list is full", msgs.UID)
if msgs.Msg[len(msgs.Msg)-1].SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) < utils.GetCurrentTimestampByMill() && msgListIsFull(msgs) {
delStruct.delUidList = append(delStruct.delUidList, msgs.UID)
lastMsgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
delStruct.minSeq = lastMsgPb.Seq + 1
log.NewDebug(operationID, utils.GetSelfFuncName(), msgs.UID, "add to delUidList", "minSeq", lastMsgPb.Seq+1)
} else {
// 列表没有满且没有不需要被删除的消息 代表他是最新的消息块
if !hasMsgDoNotNeedDel {
delStruct.minSeq, err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, len(msgs.Msg)-1)
var hasMarkDelFlag bool
for index, msg := range msgs.Msg {
if msg.SendTime == 0 {
continue
}
msgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msg.Msg, msgPb)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, "Index:", len(msgs.Msg)-1)
err = delMongoMsgsPhysical(delStruct.delUidList)
if err != nil {
return delStruct.getSetMinSeq(), err
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
if utils.GetCurrentTimestampByMill() > msg.SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) {
msgPb.Status = constant.MsgDeleted
bytes, _ := proto.Marshal(msgPb)
msgs.Msg[index].Msg = bytes
msgs.Msg[index].SendTime = 0
hasMarkDelFlag = true
} else {
if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil {
return 0, err
}
return delStruct.getSetMinSeq(), nil
if hasMarkDelFlag {
log.NewInfo(operationID, ID, "hasMarkDelFlag", "index:", index, "msgPb:", msgPb, msgs.UID)
if err := db.DB.UpdateOneMsgList(msgs); err != nil {
return delStruct.getSetMinSeq(), utils.Wrap(err, "")
}
}
return msgPb.Seq, nil
}
}
}
@@ -180,14 +167,6 @@ func msgListIsFull(chat *db.UserChat) bool {
return false
}
func CheckGroupUserMinSeq(operationID, groupID, userID string) error {
return nil
}
func CheckUserMinSeqWithMongo(operationID, userID string) error {
return nil
}
func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error {
var seqRedis uint64
var err error
internal/cron_task/clear_msg_test.go
+199 -75
View File
@@ -3,100 +3,224 @@ package cronTask
import (
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db"
"Open_IM/pkg/common/log"
pbMsg "Open_IM/pkg/proto/msg"
server_api_params "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"os/exec"
"context"
"fmt"
"strconv"
"github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"gopkg.in/mgo.v2/bson"
"testing"
"time"
)
func getMsgListFake(num int) []*pbMsg.MsgDataToMQ {
var msgList []*pbMsg.MsgDataToMQ
for i := 1; i < num; i++ {
msgList = append(msgList, &pbMsg.MsgDataToMQ{
Token: "tk",
OperationID: "operationID",
MsgData: &server_api_params.MsgData{
SendID: "sendID1",
RecvID: "recvID1",
GroupID: "",
ClientMsgID: "xxx",
ServerMsgID: "xxx",
SenderPlatformID: 1,
SenderNickname: "testNickName",
SenderFaceURL: "testFaceURL",
SessionType: 1,
MsgFrom: 100,
ContentType: 101,
Content: []byte("testFaceURL"),
Seq: uint32(i),
SendTime: time.Now().Unix(),
CreateTime: time.Now().Unix(),
Status: 1,
},
})
var (
redisClient *redis.Client
mongoClient *mongo.Collection
)
func GenUserChat(startSeq, stopSeq, delSeq, index uint32, userID string) *db.UserChat {
chat := &db.UserChat{UID: userID + ":" + strconv.Itoa(int(index))}
for i := startSeq; i <= stopSeq; i++ {
msg := server_api_params.MsgData{
SendID: "sendID1",
RecvID: "recvID1",
GroupID: "",
ClientMsgID: "xxx",
ServerMsgID: "xxx",
SenderPlatformID: 1,
SenderNickname: "testNickName",
SenderFaceURL: "testFaceURL",
SessionType: 1,
MsgFrom: 100,
ContentType: 101,
Content: []byte("testFaceURL"),
Seq: uint32(i),
SendTime: time.Now().Unix(),
CreateTime: time.Now().Unix(),
Status: 1,
}
bytes, _ := proto.Marshal(&msg)
var sendTime int64
if i <= delSeq {
sendTime = 10000
} else {
sendTime = utils.GetCurrentTimestampByMill()
}
chat.Msg = append(chat.Msg, db.MsgInfo{SendTime: int64(sendTime), Msg: bytes})
}
return msgList
return chat
}
func SetUserMaxSeq(userID string, seq int) error {
return redisClient.Set(context.Background(), "REDIS_USER_INCR_SEQ"+userID, seq, 0).Err()
}
func GetUserMinSeq(userID string) (uint64, error) {
key := "REDIS_USER_MIN_SEQ:" + userID
seq, err := redisClient.Get(context.Background(), key).Result()
return uint64(utils.StringToInt(seq)), err
}
func CreateChat(userChat *db.UserChat) error {
_, err := mongoClient.InsertOne(context.Background(), userChat)
return err
}
func DelChat(uid string, index int) error {
_, err := mongoClient.DeleteOne(context.Background(), bson.M{"uid": uid + ":" + strconv.Itoa(index)})
return err
}
func TestDeleteMongoMsgAndResetRedisSeq(t *testing.T) {
operationID := getCronTaskOperationID()
redisClient = redis.NewClient(&redis.Options{
Addr: "127.0.0.1:16379",
Password: "openIM123", // no password set
DB: 0, // use default DB
})
mongoUri := fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
"root", "openIM123", "127.0.0.1:37017",
"openIM", 100)
client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(mongoUri))
mongoClient = client.Database("openIM").Collection("msg")
testUID1 := "test_del_id1"
//testUID2 := "test_del_id2"
//testUID3 := "test_del_id3"
//testUID4 := "test_del_id4"
//testUID5 := "test_del_id5"
//testUID6 := "test_del_id6"
testUserIDList := []string{testUID1}
err := db.DB.SetUserMaxSeq(testUID1, 500)
err = db.DB.BatchInsertChat2DB(testUID1, getMsgListFake(500), testUID1+"-"+operationID, 500)
err = DelChat(testUID1, 0)
err = SetUserMaxSeq(testUID1, 600)
userChat := GenUserChat(1, 600, 200, 0, testUID1)
err = CreateChat(userChat)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID1); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID1)
}
if err := checkMaxSeqWithMongo(operationID, testUID1, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID1)
}
minSeq, err := GetUserMinSeq(testUID1)
if err != nil {
t.Error(err.Error(), testUID1)
t.Error("err is not nil", testUID1, err.Error())
}
//db.DB.SetUserMaxSeq(testUID1, 6000)
//db.DB.BatchInsertChat2DB()
//
//db.DB.SetUserMaxSeq(testUID1, 4999)
//db.DB.BatchInsertChat2DB()
//
//db.DB.SetUserMaxSeq(testUID1, 30000)
//db.DB.BatchInsertChat2DB()
//
//db.DB.SetUserMaxSeq(testUID1, 9999)
//db.DB.BatchInsertChat2DB()
cmd := exec.Command("/bin/bash", "unset $CONFIG_NAME")
_, err = cmd.StdoutPipe()
if minSeq != 201 {
t.Error("test1 is not the same", "minSeq:", minSeq, "targetSeq", 201)
}
testUID2 := "test_del_id2"
err = DelChat(testUID2, 0)
err = DelChat(testUID2, 1)
err = SetUserMaxSeq(testUID2, 7000)
userChat = GenUserChat(1, 4999, 5000, 0, testUID2)
userChat2 := GenUserChat(5000, 7000, 6000, 1, testUID2)
err = CreateChat(userChat)
err = CreateChat(userChat2)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID2); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID2)
}
if err := checkMaxSeqWithMongo(operationID, testUID2, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID2)
}
minSeq, err = GetUserMinSeq(testUID2)
if err != nil {
return
t.Error("err is not nil", testUID2, err.Error())
}
if minSeq != 6001 {
t.Error("test2 is not the same", "minSeq:", minSeq, "targetSeq", 6001)
}
//执行命令
if err := cmd.Start(); err != nil {
return
testUID3 := "test_del_id3"
err = DelChat(testUID3, 0)
err = SetUserMaxSeq(testUID3, 4999)
userChat = GenUserChat(1, 4999, 5000, 0, testUID3)
err = CreateChat(userChat)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID3); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID3)
}
for _, userID := range testUserIDList {
operationID = userID + "-" + operationID
if err := DeleteMongoMsgAndResetRedisSeq(operationID, userID); err != nil {
t.Error("checkMaxSeqWithMongo failed", userID)
}
if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", userID)
}
if err := checkMaxSeqWithMongo(operationID, testUID3, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID3)
}
minSeq, err = GetUserMinSeq(testUID3)
if err != nil {
t.Error("err is not nil", testUID3, err.Error())
}
if minSeq != 5000 {
t.Error("test3 is not the same", "minSeq:", minSeq, "targetSeq", 5000)
}
testWorkingGroupIDList := []string{"test_del_id1", "test_del_id2", "test_del_id3", "test_del_id4", "test_del_id5"}
for _, groupID := range testWorkingGroupIDList {
operationID = groupID + "-" + operationID
log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", testUserIDList)
if err := ResetUserGroupMinSeq(operationID, groupID, testUserIDList); err != nil {
t.Error("checkMaxSeqWithMongo failed", groupID)
}
if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", groupID)
}
testUID4 := "test_del_id4"
err = DelChat(testUID4, 0)
err = DelChat(testUID4, 1)
err = DelChat(testUID4, 2)
err = SetUserMaxSeq(testUID4, 12000)
userChat = GenUserChat(1, 4999, 5000, 0, testUID4)
userChat2 = GenUserChat(5000, 9999, 10000, 1, testUID4)
userChat3 := GenUserChat(10000, 12000, 11000, 2, testUID4)
err = CreateChat(userChat)
err = CreateChat(userChat2)
err = CreateChat(userChat3)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID4); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID4)
}
if err := checkMaxSeqWithMongo(operationID, testUID4, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID4)
}
minSeq, err = GetUserMinSeq(testUID4)
if err != nil {
t.Error("err is not nil", testUID4, err.Error())
}
if minSeq != 11001 {
t.Error("test4 is not the same", "minSeq:", minSeq, "targetSeq", 11001)
}
testUID5 := "test_del_id5"
err = DelChat(testUID5, 0)
err = DelChat(testUID5, 1)
err = SetUserMaxSeq(testUID5, 9999)
userChat = GenUserChat(1, 4999, 5000, 0, testUID5)
userChat2 = GenUserChat(5000, 9999, 10000, 1, testUID5)
err = CreateChat(userChat)
err = CreateChat(userChat2)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID5); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID4)
}
if err := checkMaxSeqWithMongo(operationID, testUID5, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID5)
}
minSeq, err = GetUserMinSeq(testUID5)
if err != nil {
t.Error("err is not nil", testUID5, err.Error())
}
if minSeq != 10000 {
t.Error("test5 is not the same", "minSeq:", minSeq, "targetSeq", 10000)
}
testUID6 := "test_del_id6"
err = DelChat(testUID5, 0)
err = DelChat(testUID5, 1)
err = DelChat(testUID5, 2)
err = DelChat(testUID5, 3)
userChat = GenUserChat(1, 4999, 5000, 0, testUID6)
userChat2 = GenUserChat(5000, 9999, 10000, 1, testUID6)
userChat3 = GenUserChat(10000, 14999, 13000, 2, testUID6)
userChat4 := GenUserChat(15000, 19999, 0, 3, testUID6)
err = CreateChat(userChat)
err = CreateChat(userChat2)
err = CreateChat(userChat3)
err = CreateChat(userChat4)
if err := DeleteMongoMsgAndResetRedisSeq(operationID, testUID6); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID6)
}
if err := checkMaxSeqWithMongo(operationID, testUID6, constant.WriteDiffusion); err != nil {
t.Error("checkMaxSeqWithMongo failed", testUID6)
}
minSeq, err = GetUserMinSeq(testUID6)
if err != nil {
t.Error("err is not nil", testUID6, err.Error())
}
if minSeq != 13001 {
t.Error("test3 is not the same", "minSeq:", minSeq, "targetSeq", 13001)
}
}
internal/cron_task/cron_task.go
+3 -9
View File
@@ -8,8 +8,9 @@ import (
"Open_IM/pkg/common/log"
"Open_IM/pkg/utils"
"fmt"
"github.com/robfig/cron/v3"
"time"
"github.com/robfig/cron/v3"
)
const cronTaskOperationID = "cronTaskOperationID-"
@@ -57,6 +58,7 @@ func ClearAll() {
} else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
}
// working group msg clear
workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
if err == nil {
@@ -77,9 +79,6 @@ func StartClearMsg(operationID string, userIDList []string) {
if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
}
if err := CheckUserMinSeqWithMongo(operationID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
}
}
}
@@ -98,10 +97,5 @@ func StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string)
if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
}
for _, userID := range userIDList {
if err := CheckGroupUserMinSeq(operationID, groupID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
}
}
}
}
internal/cron_task/test/main.go
-6
View File
@@ -1,6 +0,0 @@
package main
//
//func main() {
// db.DB.BatchInsertChat()
//}
internal/msg_gateway/gate/logic.go
+2
View File
@@ -186,8 +186,10 @@ func (ws *WServer) pullMsgBySeqListResp(conn *UserConn, m *Req, pb *sdk_ws.PullM
}
func (ws *WServer) userLogoutReq(conn *UserConn, m *Req) {
log.NewInfo(m.OperationID, "Ws call success to userLogoutReq start", m.SendID, m.ReqIdentifier, m.MsgIncr, string(m.Data))
rpcReq := push.DelUserPushTokenReq{}
rpcReq.UserID = m.SendID
rpcReq.PlatformID = conn.PlatformID
rpcReq.OperationID = m.OperationID
grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImPushName, m.OperationID)
if grpcConn == nil {
internal/msg_gateway/gate/relay_rpc_server.go
+49 -1
View File
@@ -225,6 +225,54 @@ func (r *RPCServer) SuperGroupOnlineBatchPushOneMsg(_ context.Context, req *pbRe
SinglePushResult: singleUserResult,
}, nil
}
func (r *RPCServer) SuperGroupBackgroundOnlinePush(_ context.Context, req *pbRelay.OnlineBatchPushOneMsgReq) (*pbRelay.OnlineBatchPushOneMsgResp, error) {
log.NewInfo(req.OperationID, "BatchPushMsgToUser is arriving", req.String())
var singleUserResult []*pbRelay.SingelMsgToUserResultList
//r.GetBatchMsgForPush(req.OperationID,req.MsgData,req.PushToUserIDList,)
msgBytes, _ := proto.Marshal(req.MsgData)
mReply := Resp{
ReqIdentifier: constant.WSPushMsg,
OperationID: req.OperationID,
Data: msgBytes,
}
var replyBytes bytes.Buffer
enc := gob.NewEncoder(&replyBytes)
err := enc.Encode(mReply)
if err != nil {
log.NewError(req.OperationID, "data encode err", err.Error())
}
for _, v := range req.PushToUserIDList {
var resp []*pbRelay.SingleMsgToUserPlatform
tempT := &pbRelay.SingelMsgToUserResultList{
UserID: v,
}
userConnMap := ws.getUserAllCons(v)
for platform, userConn := range userConnMap {
if userConn != nil && userConn.IsBackground {
temp := &pbRelay.SingleMsgToUserPlatform{
RecvID: v,
RecvPlatFormID: int32(platform),
}
if constant.PlatformIDToClass(int(userConn.PlatformID)) == constant.TerminalPC || userConn.PlatformID == constant.WebPlatformID {
resultCode := sendMsgBatchToUser(userConn, replyBytes.Bytes(), req, platform, v)
if resultCode == 0 && utils.IsContainInt(platform, r.pushTerminal) {
tempT.OnlinePush = true
promePkg.PromeInc(promePkg.MsgOnlinePushSuccessCounter)
log.Info(req.OperationID, "PushSuperMsgToUser is success By Ws", "args", req.String(), "recvPlatForm", constant.PlatformIDToName(platform), "recvID", v)
temp.ResultCode = resultCode
resp = append(resp, temp)
}
}
}
}
tempT.Resp = resp
singleUserResult = append(singleUserResult, tempT)
}
return &pbRelay.OnlineBatchPushOneMsgResp{
SinglePushResult: singleUserResult,
}, nil
}
func (r *RPCServer) OnlineBatchPushOneMsg(_ context.Context, req *pbRelay.OnlineBatchPushOneMsgReq) (*pbRelay.OnlineBatchPushOneMsgResp, error) {
log.NewInfo(req.OperationID, "BatchPushMsgToUser is arriving", req.String())
var singleUserResult []*pbRelay.SingelMsgToUserResultList
@@ -328,7 +376,7 @@ func (r *RPCServer) KickUserOffline(_ context.Context, req *pbRelay.KickUserOffl
oldConnMap := ws.getUserAllCons(v)
if conn, ok := oldConnMap[int(req.PlatformID)]; ok { // user->map[platform->conn]
log.NewWarn(req.OperationID, "send kick msg, close connection ", req.PlatformID, v)
ws.sendKickMsg(conn)
ws.sendKickMsg(conn, req.OperationID)
conn.Close()
}
}
internal/msg_gateway/gate/ws_server.go
+17 -13
View File
@@ -253,7 +253,7 @@ func (ws *WServer) MultiTerminalLoginChecker(uid string, platformID int, newConn
if oldConnMap, ok := ws.wsUserToConn[uid]; ok { // user->map[platform->conn]
if oldConn, ok := oldConnMap[platformID]; ok {
log.NewDebug(operationID, uid, platformID, "kick old conn")
ws.sendKickMsg(oldConn)
ws.sendKickMsg(oldConn, operationID)
m, err := db.DB.GetTokenMapByUidPid(uid, constant.PlatformIDToName(platformID))
if err != nil && err != go_redis.Nil {
log.NewError(operationID, "get token from redis err", err.Error(), uid, constant.PlatformIDToName(platformID))
@@ -302,11 +302,12 @@ func (ws *WServer) MultiTerminalLoginChecker(uid string, platformID int, newConn
case constant.WebAndOther:
}
}
func (ws *WServer) sendKickMsg(oldConn *UserConn) {
func (ws *WServer) sendKickMsg(oldConn *UserConn, operationID string) {
mReply := Resp{
ReqIdentifier: constant.WSKickOnlineMsg,
ErrCode: constant.ErrTokenInvalid.ErrCode,
ErrMsg: constant.ErrTokenInvalid.ErrMsg,
OperationID: operationID,
}
var b bytes.Buffer
enc := gob.NewEncoder(&b)
@@ -388,7 +389,10 @@ func (ws *WServer) delUserConn(conn *UserConn) {
if err != nil {
log.Error(operationID, " close err", "", "uid", uid, "platform", platform)
}
callbackResp := callbackUserOffline(operationID, conn.userID, platform, conn.connID)
if conn.PlatformID == 0 || conn.connID == "" {
log.NewWarn(operationID, utils.GetSelfFuncName(), "PlatformID or connID is null", conn.PlatformID, conn.connID)
}
callbackResp := callbackUserOffline(operationID, conn.userID, int(conn.PlatformID), conn.connID)
if callbackResp.ErrCode != 0 {
log.NewError(operationID, utils.GetSelfFuncName(), "callbackUserOffline failed", callbackResp)
}
@@ -419,19 +423,19 @@ func (ws *WServer) getUserAllCons(uid string) map[int]*UserConn {
return nil
}
//func (ws *WServer) getUserUid(conn *UserConn) (uid string, platform int) {
// rwLock.RLock()
// defer rwLock.RUnlock()
// func (ws *WServer) getUserUid(conn *UserConn) (uid string, platform int) {
// rwLock.RLock()
// defer rwLock.RUnlock()
//
// if stringMap, ok := ws.wsConnToUser[conn]; ok {
// for k, v := range stringMap {
// platform = k
// uid = v
// if stringMap, ok := ws.wsConnToUser[conn]; ok {
// for k, v := range stringMap {
// platform = k
// uid = v
// }
// return uid, platform
// }
// return uid, platform
// return "", 0
// }
// return "", 0
//}
func (ws *WServer) headerCheck(w http.ResponseWriter, r *http.Request, operationID string) (isPass, compression bool) {
status := http.StatusUnauthorized
query := r.URL.Query()
internal/msg_transfer/logic/init.go
+5
View File
@@ -22,7 +22,9 @@ var (
persistentCH PersistentConsumerHandler
historyCH OnlineHistoryRedisConsumerHandler
historyMongoCH OnlineHistoryMongoConsumerHandler
modifyCH ModifyMsgConsumerHandler
producer *kafka.Producer
producerToModify *kafka.Producer
producerToMongo *kafka.Producer
cmdCh chan Cmd2Value
onlineTopicStatus int
@@ -43,11 +45,13 @@ func Init() {
persistentCH.Init() // ws2mschat save mysql
historyCH.Init(cmdCh) //
historyMongoCH.Init()
modifyCH.Init()
onlineTopicStatus = OnlineTopicVacancy
//offlineHistoryCH.Init(cmdCh)
statistics.NewStatistics(&singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
statistics.NewStatistics(&groupMsgCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second groupMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
producer = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
producerToModify = kafka.NewKafkaProducer(config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.MsgToModify.Topic)
producerToMongo = kafka.NewKafkaProducer(config.Config.Kafka.MsgToMongo.Addr, config.Config.Kafka.MsgToMongo.Topic)
}
func Run(promethuesPort int) {
@@ -59,6 +63,7 @@ func Run(promethuesPort int) {
}
go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
go historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyMongoCH)
go modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(&modifyCH)
//go offlineHistoryCH.historyConsumerGroup.RegisterHandleAndConsumer(&offlineHistoryCH)
go func() {
err := promePkg.StartPromeSrv(promethuesPort)
internal/msg_transfer/logic/modify_msg_handler.go
+121
View File
@@ -0,0 +1,121 @@
package logic
import (
"Open_IM/pkg/base_info"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db"
kfk "Open_IM/pkg/common/kafka"
"Open_IM/pkg/common/log"
pbMsg "Open_IM/pkg/proto/msg"
server_api_params "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"encoding/json"
"github.com/Shopify/sarama"
"github.com/golang/protobuf/proto"
)
type ModifyMsgConsumerHandler struct {
msgHandle map[string]fcb
modifyMsgConsumerGroup *kfk.MConsumerGroup
}
func (mmc *ModifyMsgConsumerHandler) Init() {
mmc.msgHandle = make(map[string]fcb)
mmc.msgHandle[config.Config.Kafka.MsgToModify.Topic] = mmc.ModifyMsg
mmc.modifyMsgConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify)
}
func (ModifyMsgConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (ModifyMsgConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
if len(msg.Value) != 0 {
mmc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
} else {
log.Error("", "msg get from kafka but is nil", msg.Key)
}
sess.MarkMessage(msg, "")
}
return nil
}
func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
log.NewInfo("msg come here ModifyMsg!!!", "", "msg", string(cMsg.Value), msgKey)
msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
if err != nil {
log.NewError(msgFromMQ.TriggerID, "msg_transfer Unmarshal msg err", "msg", string(cMsg.Value), "err", err.Error())
return
}
log.Debug(msgFromMQ.TriggerID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
for _, msgDataToMQ := range msgFromMQ.MessageList {
isReactionFromCache := utils.GetSwitchFromOptions(msgDataToMQ.MsgData.Options, constant.IsReactionFromCache)
if !isReactionFromCache {
continue
}
if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier {
notification := &base_info.ReactionMessageModifierNotification{}
if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
continue
}
if notification.IsExternalExtensions {
log.NewInfo(msgDataToMQ.OperationID, "msg:", notification, "this is external extensions")
continue
}
if !notification.IsReact {
// first time to modify
var reactionExtensionList = make(map[string]db.KeyValue)
extendMsg := db.ExtendMsg{
ReactionExtensionList: reactionExtensionList,
ClientMsgID: notification.ClientMsgID,
MsgFirstModifyTime: notification.MsgFirstModifyTime,
}
for _, v := range notification.SuccessReactionExtensionList {
reactionExtensionList[v.TypeKey] = db.KeyValue{
TypeKey: v.TypeKey,
Value: v.Value,
LatestUpdateTime: v.LatestUpdateTime,
}
}
if err := db.DB.InsertExtendMsg(notification.SourceID, notification.SessionType, &extendMsg); err != nil {
log.NewError(msgDataToMQ.OperationID, "MsgFirstModify InsertExtendMsg failed", notification.SourceID, notification.SessionType, extendMsg, err.Error())
continue
}
} else {
var reactionExtensionList = make(map[string]*server_api_params.KeyValue)
for _, v := range notification.SuccessReactionExtensionList {
reactionExtensionList[v.TypeKey] = &server_api_params.KeyValue{
TypeKey: v.TypeKey,
Value: v.Value,
LatestUpdateTime: v.LatestUpdateTime,
}
}
// is already modify
if err := db.DB.InsertOrUpdateReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
} else if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageDeleter {
notification := &base_info.ReactionMessageDeleteNotification{}
if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
continue
}
if err := db.DB.DeleteReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil {
log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
}
}
func UnMarshallSetReactionMsgContent(content []byte) (notification *base_info.ReactionMessageModifierNotification, err error) {
return notification, nil
}
internal/msg_transfer/logic/online_history_msg_handler.go
+19 -2
View File
@@ -72,6 +72,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
storageMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
notStoragePushMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
log.Debug(triggerID, "msg arrived channel", "channel id", channelID, msgList, msgChannelValue.aggregationID, len(msgList))
var modifyMsgList []*pbMsg.MsgDataToMQ
for _, v := range msgList {
log.Debug(triggerID, "msg come to storage center", v.String())
isHistory := utils.GetSwitchFromOptions(v.MsgData.Options, constant.IsHistory)
@@ -83,11 +84,15 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
if !(!isSenderSync && msgChannelValue.aggregationID == v.MsgData.SendID) {
notStoragePushMsgList = append(notStoragePushMsgList, v)
}
}
if v.MsgData.ContentType == constant.ReactionMessageModifier || v.MsgData.ContentType == constant.ReactionMessageDeleter {
modifyMsgList = append(modifyMsgList, v)
}
}
if len(modifyMsgList) > 0 {
sendMessageToModifyMQ(msgChannelValue.aggregationID, triggerID, modifyMsgList)
}
//switch msgChannelValue.msg.MsgData.SessionType {
//case constant.SingleChatType:
//case constant.GroupChatType:
@@ -107,6 +112,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
singleMsgSuccessCount += uint64(len(storageMsgList))
singleMsgSuccessCountMutex.Unlock()
och.SendMessageToMongoCH(msgChannelValue.aggregationID, triggerID, storageMsgList, lastSeq)
for _, v := range storageMsgList {
sendMessageToPushMQ(v, msgChannelValue.aggregationID)
}
@@ -552,6 +558,17 @@ func sendMessageToPushMQ(message *pbMsg.MsgDataToMQ, pushToUserID string) {
return
}
func sendMessageToModifyMQ(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ) {
if len(messages) > 0 {
pid, offset, err := producerToModify.SendMessage(&pbMsg.MsgDataToModifyByMQ{AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
if err != nil {
log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
} else {
// log.NewWarn(m.OperationID, "sendMsgToKafka client msgID ", m.MsgData.ClientMsgID)
}
}
}
// String hashes a string to a unique hashcode.
//
// crc32 returns a uint32, but for our use we need
internal/msg_transfer/logic/online_msg_to_mongo_handler.go
-1
View File
@@ -61,7 +61,6 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
}
//if v.MsgData.ContentType == ? {}
}
}
}
internal/push/getui/push.go
+118 -52
View File
@@ -26,8 +26,10 @@ var (
)
const (
PushURL = "/push/single/alias"
AuthURL = "/auth"
PushURL = "/push/single/alias"
AuthURL = "/auth"
TaskURL = "/push/list/message"
BatchPushURL = "/push/list/alias"
)
func init() {
@@ -53,23 +55,42 @@ type AuthResp struct {
Token string `json:"token"`
}
type TaskResp struct {
TaskID string `json:"taskID"`
}
type Settings struct {
TTL *int64 `json:"ttl"`
}
type Audience struct {
Alias []string `json:"alias"`
}
type PushMessage struct {
Notification *Notification `json:"notification,omitempty"`
Transmission *string `json:"transmission,omitempty"`
}
type PushChannel struct {
Ios *Ios `json:"ios"`
Android *Android `json:"android"`
}
type PushReq struct {
RequestID string `json:"request_id"`
Audience struct {
Alias []string `json:"alias"`
} `json:"audience"`
PushMessage struct {
Notification Notification `json:"notification,omitempty"`
Transmission string `json:"transmission,omitempty"`
} `json:"push_message"`
PushChannel struct {
Ios Ios `json:"ios"`
Android Android `json:"android"`
} `json:"push_channel"`
RequestID *string `json:"request_id"`
Settings *Settings `json:"settings"`
Audience *Audience `json:"audience"`
PushMessage *PushMessage `json:"push_message"`
PushChannel *PushChannel `json:"push_channel"`
IsAsync *bool `json:"is_async"`
Taskid *string `json:"taskid"`
}
type Ios struct {
Aps struct {
NotiType *string `json:"type"`
AutoBadge *string `json:"auto_badge"`
Aps struct {
Sound string `json:"sound"`
Alert Alert `json:"alert"`
} `json:"aps"`
@@ -119,9 +140,9 @@ func newGetuiClient() *Getui {
func (g *Getui) Push(userIDList []string, title, detailContent, operationID string, opts push.PushOpts) (resp string, err error) {
token, err := db.DB.GetGetuiToken()
log.NewDebug(operationID, utils.GetSelfFuncName(), "token：", token)
log.NewDebug(operationID, utils.GetSelfFuncName(), "token：", token, userIDList)
if err != nil {
log.NewError(operationID, utils.OperationIDGenerator(), "GetGetuiToken failed", err.Error())
log.NewError(operationID, utils.GetSelfFuncName(), "GetGetuiToken failed", err.Error())
}
if token == "" || err != nil {
token, err = g.getTokenAndSave2Redis(operationID)
@@ -130,47 +151,32 @@ func (g *Getui) Push(userIDList []string, title, detailContent, operationID stri
return "", utils.Wrap(err, "")
}
}
pushReq := PushReq{
RequestID: utils.OperationIDGenerator(),
Audience: struct {
Alias []string `json:"alias"`
}{Alias: []string{userIDList[0]}},
}
pushReq.PushMessage.Notification = Notification{
pushReq := PushReq{PushMessage: &PushMessage{Notification: &Notification{
Title: title,
Body: detailContent,
ClickType: "startapp",
ChannelID: config.Config.Push.Getui.ChannelID,
ChannelName: config.Config.Push.Getui.ChannelName,
}
pushReq.PushChannel.Ios.Aps.Sound = "default"
pushReq.PushChannel.Ios.Aps.Alert = Alert{
Title: title,
Body: title,
}
pushReq.PushChannel.Android.Ups.Notification = Notification{
Title: title,
Body: title,
ClickType: "startapp",
}
pushReq.PushChannel.Android.Ups.Options = Options{
HW: struct {
DefaultSound bool `json:"/message/android/notification/default_sound"`
ChannelID string `json:"/message/android/notification/channel_id"`
Sound string `json:"/message/android/notification/sound"`
Importance string `json:"/message/android/notification/importance"`
}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL"},
XM: struct {
ChannelID string `json:"/extra.channel_id"`
}{ChannelID: "high_system"},
VV: struct {
Classification int "json:\"/classification\""
}{
Classification: 1,
},
}
}}}
pushReq.setPushChannel(title, detailContent)
pushResp := PushResp{}
err = g.request(PushURL, pushReq, token, &pushResp, operationID)
if len(userIDList) > 1 {
taskID, err := g.GetTaskID(operationID, token, pushReq)
if err != nil {
return "", utils.Wrap(err, "GetTaskIDAndSave2Redis failed")
}
pushReq = PushReq{Audience: &Audience{Alias: userIDList}}
var IsAsync = true
pushReq.IsAsync = &IsAsync
pushReq.Taskid = &taskID
err = g.request(BatchPushURL, pushReq, token, &pushResp, operationID)
} else {
reqID := utils.OperationIDGenerator()
pushReq.RequestID = &reqID
pushReq.Audience = &Audience{Alias: []string{userIDList[0]}}
err = g.request(PushURL, pushReq, token, &pushResp, operationID)
}
switch err {
case TokenExpireError:
token, err = g.getTokenAndSave2Redis(operationID)
@@ -209,6 +215,17 @@ func (g *Getui) Auth(operationID string, timeStamp int64) (token string, expireT
return respAuth.Token, int64(expire), err
}
func (g *Getui) GetTaskID(operationID, token string, pushReq PushReq) (string, error) {
respTask := TaskResp{}
ttl := int64(1000 * 60 * 5)
pushReq.Settings = &Settings{TTL: &ttl}
err := g.request(TaskURL, pushReq, token, &respTask, operationID)
if err != nil {
return "", utils.Wrap(err, "")
}
return respTask.TaskID, nil
}
func (g *Getui) request(url string, content interface{}, token string, returnStruct interface{}, operationID string) error {
con, err := json.Marshal(content)
if err != nil {
@@ -245,6 +262,41 @@ func (g *Getui) request(url string, content interface{}, token string, returnStr
return nil
}
func (pushReq *PushReq) setPushChannel(title string, body string) {
pushReq.PushChannel = &PushChannel{}
autoBadge := "+1"
pushReq.PushChannel.Ios = &Ios{AutoBadge: &autoBadge}
notify := "notify"
pushReq.PushChannel.Ios.NotiType = &notify
pushReq.PushChannel.Ios.Aps.Sound = "default"
pushReq.PushChannel.Ios.Aps.Alert = Alert{
Title: title,
Body: body,
}
pushReq.PushChannel.Android = &Android{}
pushReq.PushChannel.Android.Ups.Notification = Notification{
Title: title,
Body: body,
ClickType: "startapp",
}
pushReq.PushChannel.Android.Ups.Options = Options{
HW: struct {
DefaultSound bool `json:"/message/android/notification/default_sound"`
ChannelID string `json:"/message/android/notification/channel_id"`
Sound string `json:"/message/android/notification/sound"`
Importance string `json:"/message/android/notification/importance"`
}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL"},
XM: struct {
ChannelID string `json:"/extra.channel_id"`
}{ChannelID: "high_system"},
VV: struct {
Classification int "json:\"/classification\""
}{
Classification: 1,
},
}
}
func (g *Getui) getTokenAndSave2Redis(operationID string) (token string, err error) {
token, expireTime, err := g.Auth(operationID, time.Now().UnixNano()/1e6)
if err != nil {
@@ -257,3 +309,17 @@ func (g *Getui) getTokenAndSave2Redis(operationID string) (token string, err err
}
return token, nil
}
func (g *Getui) GetTaskIDAndSave2Redis(operationID, token string, pushReq PushReq) (taskID string, err error) {
ttl := int64(1000 * 60 * 60 * 24)
pushReq.Settings = &Settings{TTL: &ttl}
taskID, err = g.GetTaskID(operationID, token, pushReq)
if err != nil {
return "", utils.Wrap(err, "GetTaskIDAndSave2Redis failed")
}
err = db.DB.SetGetuiTaskID(taskID, 60*60*23)
if err != nil {
return "", utils.Wrap(err, "Auth failed")
}
return token, nil
}
internal/push/logic/push_rpc_server.go
+2 -1
View File
@@ -93,10 +93,11 @@ func (r *RPCServer) PushMsg(_ context.Context, pbData *pbPush.PushMsgReq) (*pbPu
}
func (r *RPCServer) DelUserPushToken(c context.Context, req *pbPush.DelUserPushTokenReq) (*pbPush.DelUserPushTokenResp, error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "req", req.String())
var resp pbPush.DelUserPushTokenResp
err := db.DB.DelFcmToken(req.UserID, int(req.PlatformID))
if err != nil {
errMsg := req.OperationID + " " + "SetFcmToken failed " + err.Error()
errMsg := req.OperationID + " " + "DelFcmToken failed " + err.Error()
log.NewError(req.OperationID, errMsg)
resp.ErrCode = 500
resp.ErrMsg = errMsg
internal/push/logic/push_to_client.go
+41 -1
View File
@@ -210,11 +210,25 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
successCount++
if isOfflinePush {
var onlineSuccessUserIDList []string
var WebAndPcBackgroundUserIDList []string
onlineSuccessUserIDList = append(onlineSuccessUserIDList, pushMsg.MsgData.SendID)
for _, v := range wsResult {
if v.OnlinePush && v.UserID != pushMsg.MsgData.SendID {
onlineSuccessUserIDList = append(onlineSuccessUserIDList, v.UserID)
}
if !v.OnlinePush {
if len(v.Resp) != 0 {
for _, singleResult := range v.Resp {
if singleResult.ResultCode == -2 {
if constant.PlatformIDToClass(int(singleResult.RecvPlatFormID)) == constant.TerminalPC ||
singleResult.RecvPlatFormID == constant.WebPlatformID {
WebAndPcBackgroundUserIDList = append(WebAndPcBackgroundUserIDList, v.UserID)
}
}
}
}
}
}
onlineFailedUserIDList := utils.DifferenceString(onlineSuccessUserIDList, pushToUserIDList)
//Use offline push messaging
@@ -240,7 +254,17 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
} else {
needOfflinePushUserIDList = onlineFailedUserIDList
}
if pushMsg.MsgData.ContentType != constant.SignalingNotification {
notNotificationUserIDList, err := db.DB.GetSuperGroupUserReceiveNotNotifyMessageIDList(pushMsg.MsgData.GroupID)
if err != nil {
log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "GetSuperGroupUserReceiveNotNotifyMessageIDList failed", pushMsg.MsgData.GroupID)
} else {
log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), notNotificationUserIDList)
}
needOfflinePushUserIDList = utils.RemoveFromSlice(notNotificationUserIDList, needOfflinePushUserIDList)
log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), needOfflinePushUserIDList)
}
if offlinePusher == nil {
return
}
@@ -248,7 +272,7 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
if err != nil {
log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "GetOfflinePushOpts failed", pushMsg, err.Error())
}
log.NewInfo(pushMsg.OperationID, utils.GetSelfFuncName(), onlineFailedUserIDList, title, detailContent, "opts:", opts)
log.NewInfo(pushMsg.OperationID, utils.GetSelfFuncName(), needOfflinePushUserIDList, title, detailContent, "opts:", opts)
if title == "" {
switch pushMsg.MsgData.ContentType {
case constant.Text:
@@ -285,6 +309,22 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
promePkg.PromeInc(promePkg.MsgOfflinePushSuccessCounter)
log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
}
needBackgroupPushUserID := utils.IntersectString(needOfflinePushUserIDList, WebAndPcBackgroundUserIDList)
grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), pushMsg.OperationID)
if len(needBackgroupPushUserID) > 0 {
//Online push message
log.Debug(pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String())
for _, v := range grpcCons {
msgClient := pbRelay.NewRelayClient(v)
_, err := msgClient.SuperGroupBackgroundOnlinePush(context.Background(), &pbRelay.OnlineBatchPushOneMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData,
PushToUserIDList: needBackgroupPushUserID})
if err != nil {
log.NewError("push data to client rpc err", pushMsg.OperationID, "err", err)
continue
}
}
}
}
}
}
internal/push/push_interface.go
+1 -1
View File
@@ -2,7 +2,7 @@ package push
import "Open_IM/pkg/common/constant"
var PushTerminal = []int{constant.IOSPlatformID, constant.AndroidPlatformID}
var PushTerminal = []int{constant.IOSPlatformID, constant.AndroidPlatformID, constant.WebPlatformID}
type OfflinePusher interface {
Push(userIDList []string, title, detailContent, operationID string, opts PushOpts) (resp string, err error)
internal/rpc/conversation/conversaion.go
+15
View File
@@ -61,6 +61,21 @@ func (rpc *rpcConversation) ModifyConversationField(c context.Context, req *pbCo
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
if req.Conversation.ConversationType == constant.SuperGroupChatType {
if req.Conversation.RecvMsgOpt == constant.ReceiveNotNotifyMessage {
if err = db.DB.SetSuperGroupUserReceiveNotNotifyMessage(req.Conversation.GroupID, v); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), req.Conversation.GroupID, v)
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
} else {
if err = db.DB.SetSuperGroupUserReceiveNotifyMessage(req.Conversation.GroupID, v); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), req.Conversation.GroupID, v)
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
}
}
}
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"recv_msg_opt": conversation.RecvMsgOpt})
case constant.FieldGroupAtType:
internal/rpc/group/callback.go
+55
View File
@@ -10,6 +10,8 @@ import (
pbGroup "Open_IM/pkg/proto/group"
"Open_IM/pkg/utils"
http2 "net/http"
"google.golang.org/protobuf/types/known/wrapperspb"
)
func callbackBeforeCreateGroup(req *pbGroup.CreateGroupReq) cbApi.CommonCallbackResp {
@@ -126,3 +128,56 @@ func CallbackBeforeMemberJoinGroup(operationID string, groupMember *db.GroupMemb
}
return callbackResp
}
func CallbackBeforeSetGroupMemberInfo(req *pbGroup.SetGroupMemberInfoReq) cbApi.CommonCallbackResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: req.OperationID}
if !config.Config.Callback.CallbackBeforeSetGroupMemberInfo.Enable {
return callbackResp
}
callbackReq := cbApi.CallbackBeforeSetGroupMemberInfoReq{
CallbackCommand: constant.CallbackBeforeSetGroupMemberInfoCommand,
OperationID: req.OperationID,
GroupID: req.GroupID,
UserID: req.UserID,
}
if req.Nickname != nil {
callbackReq.Nickname = req.Nickname.Value
}
if req.FaceURL != nil {
callbackReq.FaceURL = req.FaceURL.Value
}
if req.RoleLevel != nil {
callbackReq.RoleLevel = req.RoleLevel.Value
}
if req.Ex != nil {
callbackReq.Ex = req.Ex.Value
}
resp := &cbApi.CallbackBeforeSetGroupMemberInfoResp{
CommonCallbackResp: &callbackResp,
}
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetGroupMemberInfoCommand, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo.CallbackTimeOut); err != nil {
callbackResp.ErrCode = http2.StatusInternalServerError
callbackResp.ErrMsg = err.Error()
if !config.Config.Callback.CallbackBeforeSetGroupMemberInfo.CallbackFailedContinue {
callbackResp.ActionCode = constant.ActionForbidden
return callbackResp
} else {
callbackResp.ActionCode = constant.ActionAllow
return callbackResp
}
}
if resp.FaceURL != nil {
req.FaceURL = &wrapperspb.StringValue{Value: *resp.FaceURL}
}
if resp.Nickname != nil {
req.Nickname = &wrapperspb.StringValue{Value: *resp.Nickname}
}
if resp.RoleLevel != nil {
req.RoleLevel = &wrapperspb.Int32Value{Value: *resp.RoleLevel}
}
if resp.Ex != nil {
req.Ex = &wrapperspb.StringValue{Value: *resp.Ex}
}
return callbackResp
}
internal/rpc/group/group.go
+161 -7
View File
@@ -30,6 +30,7 @@ import (
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/wrapperspb"
"gorm.io/gorm"
)
@@ -364,6 +365,14 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
var resp pbGroup.InviteUserToGroupResp
joinReq := pbGroup.JoinGroupReq{}
for _, v := range req.InvitedUserIDList {
if imdb.IsExistGroupMember(req.GroupID, v) {
log.NewError(req.OperationID, "IsExistGroupMember ", req.GroupID, v)
var resultNode pbGroup.Id2Result
resultNode.Result = -1
resultNode.UserID = v
resp.Id2ResultList = append(resp.Id2ResultList, &resultNode)
continue
}
var groupRequest db.GroupRequest
groupRequest.UserID = v
groupRequest.GroupID = req.GroupID
@@ -451,8 +460,19 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
resp.Id2ResultList = append(resp.Id2ResultList, &resultNode)
}
} else {
okUserIDList = req.InvitedUserIDList
if err := db.DB.AddUserToSuperGroup(req.GroupID, req.InvitedUserIDList); err != nil {
for _, v := range req.InvitedUserIDList {
if imdb.IsExistGroupMember(req.GroupID, v) {
log.NewError(req.OperationID, "IsExistGroupMember ", req.GroupID, v)
var resultNode pbGroup.Id2Result
resultNode.Result = -1
resp.Id2ResultList = append(resp.Id2ResultList, &resultNode)
continue
} else {
okUserIDList = append(okUserIDList, v)
}
}
//okUserIDList = req.InvitedUserIDList
if err := db.DB.AddUserToSuperGroup(req.GroupID, okUserIDList); err != nil {
log.NewError(req.OperationID, "AddUserToSuperGroup failed ", req.GroupID, err)
return &pbGroup.InviteUserToGroupResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: err.Error()}, nil
}
@@ -723,6 +743,33 @@ func (s *groupServer) KickGroupMember(ctx context.Context, req *pbGroup.KickGrou
resp.ErrMsg = constant.ErrDB.ErrMsg
return &resp, nil
}
if err := rocksCache.DelGroupMemberListHashFromCache(req.GroupID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), req.GroupID, err.Error())
}
if err := rocksCache.DelGroupMemberIDListFromCache(req.GroupID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), req.GroupID)
}
reqPb := pbConversation.ModifyConversationFieldReq{Conversation: &pbConversation.Conversation{}}
reqPb.OperationID = req.OperationID
reqPb.UserIDList = okUserIDList
reqPb.FieldType = constant.FieldUnread
reqPb.Conversation.GroupID = req.GroupID
reqPb.Conversation.ConversationID = utils.GetConversationIDBySessionType(req.GroupID, constant.SuperGroupChatType)
reqPb.Conversation.ConversationType = int32(constant.SuperGroupChatType)
reqPb.Conversation.UpdateUnreadCountTime = utils.GetCurrentTimestampByMill()
etcdConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName, req.OperationID)
if etcdConn == nil {
errMsg := req.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
}
client := pbConversation.NewConversationClient(etcdConn)
respPb, err := client.ModifyConversationField(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "ModifyConversationField rpc failed, ", reqPb.String(), err.Error())
} else {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "ModifyConversationField success", respPb.String())
}
}
if groupInfo.GroupType != constant.SuperGroup {
@@ -756,7 +803,15 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbGroup.GetG
var resp pbGroup.GetGroupMembersInfoResp
resp.MemberList = []*open_im_sdk.GroupMemberFullInfo{}
for _, userID := range req.MemberList {
groupMember, err := rocksCache.GetGroupMemberInfoFromCache(req.GroupID, userID)
var (
groupMember *db.GroupMember
err error
)
if req.NoCache {
groupMember, err = imdb.GetGroupMemberInfoByGroupIDAndUserID(req.GroupID, userID)
} else {
groupMember, err = rocksCache.GetGroupMemberInfoFromCache(req.GroupID, userID)
}
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), req.GroupID, userID, err.Error())
continue
@@ -861,6 +916,10 @@ func (s *groupServer) GroupApplicationResponse(_ context.Context, req *pbGroup.G
log.NewError(req.OperationID, "GroupApplicationResponse failed ", err.Error(), req.FromUserID)
return &pbGroup.GroupApplicationResponseResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil
}
if imdb.IsExistGroupMember(req.GroupID, req.FromUserID) {
log.NewInfo(req.OperationID, "GroupApplicationResponse user in group", req.GroupID, req.FromUserID)
return &pbGroup.GroupApplicationResponseResp{CommonResp: &pbGroup.CommonResp{}}, nil
}
member := db.GroupMember{}
member.GroupID = req.GroupID
member.UserID = req.FromUserID
@@ -976,12 +1035,15 @@ func (s *groupServer) GroupApplicationResponse(_ context.Context, req *pbGroup.G
func (s *groupServer) JoinGroup(ctx context.Context, req *pbGroup.JoinGroupReq) (*pbGroup.JoinGroupResp, error) {
log.NewInfo(req.OperationID, "JoinGroup args ", req.String())
if imdb.IsExistGroupMember(req.GroupID, req.OpUserID) {
log.NewInfo(req.OperationID, "IsExistGroupMember", req.GroupID, req.OpUserID)
return &pbGroup.JoinGroupResp{CommonResp: &pbGroup.CommonResp{}}, nil
}
_, err := imdb.GetUserByUserID(req.OpUserID)
if err != nil {
log.NewError(req.OperationID, "GetUserByUserID failed ", err.Error(), req.OpUserID)
return &pbGroup.JoinGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil
}
groupInfo, err := rocksCache.GetGroupInfoFromCache(req.GroupID)
if err != nil {
log.NewError(req.OperationID, "GetGroupInfoByGroupID failed ", req.GroupID, err)
@@ -1274,7 +1336,8 @@ func (s *groupServer) SetGroupInfo(ctx context.Context, req *pbGroup.SetGroupInf
}
log.NewInfo(req.OperationID, "SetGroupInfo rpc return ", pbGroup.SetGroupInfoResp{CommonResp: &pbGroup.CommonResp{}})
if changedType != 0 {
chat.GroupInfoSetNotification(req.OperationID, req.OpUserID, req.GroupInfoForSet.GroupID, groupName, notification, introduction, faceURL, req.GroupInfoForSet.NeedVerification)
chat.GroupInfoSetNotification(req.OperationID, req.OpUserID, req.GroupInfoForSet.GroupID, groupName, notification,
introduction, faceURL, req.GroupInfoForSet.NeedVerification, req.GroupInfoForSet.ApplyMemberFriend, req.GroupInfoForSet.LookMemberInfo)
}
if req.GroupInfoForSet.Notification != "" {
//get group member user id
@@ -1765,11 +1828,35 @@ func (s *groupServer) SetGroupMemberNickname(ctx context.Context, req *pbGroup.S
log.Error(req.OperationID, errMsg)
return &pbGroup.SetGroupMemberNicknameResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrAccess.ErrCode, ErrMsg: constant.ErrAccess.ErrMsg}}, nil
}
cbReq := &pbGroup.SetGroupMemberInfoReq{
GroupID: req.GroupID,
UserID: req.UserID,
OperationID: req.OperationID,
OpUserID: req.OpUserID,
Nickname: &wrapperspb.StringValue{Value: req.Nickname},
}
callbackResp := CallbackBeforeSetGroupMemberInfo(cbReq)
if callbackResp.ErrCode != 0 {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup resp: ", callbackResp)
}
if callbackResp.ActionCode != constant.ActionAllow {
if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201
}
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup result", "end rpc and return", callbackResp)
return &pbGroup.SetGroupMemberNicknameResp{
CommonResp: &pbGroup.CommonResp{
ErrCode: int32(callbackResp.ErrCode),
ErrMsg: callbackResp.ErrMsg,
},
}, nil
}
nickName := cbReq.Nickname.Value
groupMemberInfo := db.GroupMember{}
groupMemberInfo.UserID = req.UserID
groupMemberInfo.GroupID = req.GroupID
if req.Nickname == "" {
if nickName == "" {
userNickname, err := imdb.GetUserNameByUserID(groupMemberInfo.UserID)
if err != nil {
errMsg := req.OperationID + " GetUserNameByUserID failed " + err.Error()
@@ -1778,7 +1865,7 @@ func (s *groupServer) SetGroupMemberNickname(ctx context.Context, req *pbGroup.S
}
groupMemberInfo.Nickname = userNickname
} else {
groupMemberInfo.Nickname = req.Nickname
groupMemberInfo.Nickname = nickName
}
if err := rocksCache.DelGroupMemberInfoFromCache(req.GroupID, req.UserID); err != nil {
@@ -1805,6 +1892,23 @@ func (s *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbGroup.SetGr
resp.CommonResp.ErrMsg = err.Error()
return resp, nil
}
callbackResp := CallbackBeforeSetGroupMemberInfo(req)
if callbackResp.ErrCode != 0 {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup resp: ", callbackResp)
}
if callbackResp.ActionCode != constant.ActionAllow {
if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201
}
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup result", "end rpc and return", callbackResp)
return &pbGroup.SetGroupMemberInfoResp{
CommonResp: &pbGroup.CommonResp{
ErrCode: int32(callbackResp.ErrCode),
ErrMsg: callbackResp.ErrMsg,
},
}, nil
}
groupMember := db.GroupMember{
GroupID: req.GroupID,
UserID: req.UserID,
@@ -1913,3 +2017,53 @@ func (s *groupServer) DelGroupAndUserCache(operationID, groupID string, userIDLi
}
return nil
}
func (s *groupServer) GroupIsExist(c context.Context, req *pbGroup.GroupIsExistReq) (*pbGroup.GroupIsExistResp, error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp := &pbGroup.GroupIsExistResp{CommonResp: &pbGroup.CommonResp{}}
groups, err := imdb.GetGroupInfoByGroupIDList(req.GroupIDList)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), "args:", req.GroupIDList)
resp.CommonResp.ErrMsg = err.Error()
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
return resp, nil
}
var m = make(map[string]bool)
for _, groupID := range req.GroupIDList {
m[groupID] = false
for _, group := range groups {
if groupID == group.GroupID {
m[groupID] = true
break
}
}
}
resp.IsExistMap = m
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", req.String())
return resp, nil
}
func (s *groupServer) UserIsInGroup(c context.Context, req *pbGroup.UserIsInGroupReq) (*pbGroup.UserIsInGroupResp, error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp := &pbGroup.UserIsInGroupResp{}
groupMemberList, err := imdb.GetGroupMemberByUserIDList(req.GroupID, req.UserIDList)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), "args:", req.GroupID, req.UserIDList)
resp.CommonResp.ErrMsg = err.Error()
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
return resp, nil
}
var m = make(map[string]bool)
for _, userID := range req.UserIDList {
m[userID] = false
for _, user := range groupMemberList {
if userID == user.UserID {
m[userID] = true
break
}
}
}
resp.IsExistMap = m
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", req.String())
return resp, nil
}
internal/rpc/msg/callback.go
-6
View File
@@ -221,9 +221,3 @@ func callbackMsgModify(msg *pbChat.SendMsgReq) cbApi.CommonCallbackResp {
log.NewDebug(msg.OperationID, utils.GetSelfFuncName(), string(msg.MsgData.Content))
return callbackResp
}
func CallbackBeforeExtendMsgModify() cbApi.CommonCallbackResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: ""}
return callbackResp
}
internal/rpc/msg/del_msg.go
+5 -5
View File
@@ -5,8 +5,8 @@ import (
"Open_IM/pkg/common/db"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/token_verify"
commonPb "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/proto/msg"
commonPb "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"context"
"time"
@@ -39,18 +39,18 @@ func (rpc *rpcChat) DelSuperGroupMsg(_ context.Context, req *msg.DelSuperGroupMs
resp := &msg.DelSuperGroupMsgResp{}
groupMaxSeq, err := db.DB.GetGroupMaxSeq(req.GroupID)
if err != nil {
log.NewError(req.OperationID, "GetGroupMaxSeq false ", req.OpUserID, req.UserID,req.GroupID)
log.NewError(req.OperationID, "GetGroupMaxSeq false ", req.OpUserID, req.UserID, req.GroupID)
resp.ErrCode = constant.ErrDB.ErrCode
resp.ErrMsg = err.Error()
return resp, nil
}
err = db.DB.SetGroupUserMinSeq(req.GroupID,req.UserID, groupMaxSeq)
err = db.DB.SetGroupUserMinSeq(req.GroupID, req.UserID, groupMaxSeq+1)
if err != nil {
log.NewError(req.OperationID, "SetGroupUserMinSeq false ", req.OpUserID, req.UserID,req.GroupID)
log.NewError(req.OperationID, "SetGroupUserMinSeq false ", req.OpUserID, req.UserID, req.GroupID)
resp.ErrCode = constant.ErrDB.ErrCode
resp.ErrMsg = err.Error()
return resp, nil
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
}
}
internal/rpc/msg/extend_msg.go
+476
View File
@@ -1 +1,477 @@
package msg
import (
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db"
"Open_IM/pkg/common/log"
"Open_IM/pkg/proto/msg"
"Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"context"
go_redis "github.com/go-redis/redis/v8"
"time"
)
func (rpc *rpcChat) SetMessageReactionExtensions(ctx context.Context, req *msg.SetMessageReactionExtensionsReq) (resp *msg.SetMessageReactionExtensionsResp, err error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
var rResp msg.SetMessageReactionExtensionsResp
rResp.ClientMsgID = req.ClientMsgID
rResp.MsgFirstModifyTime = req.MsgFirstModifyTime
callbackResp := 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
}
//if ExternalExtension
if req.IsExternalExtensions {
var isHistory bool
if req.IsReact {
isHistory = false
} else {
isHistory = true
}
rResp.MsgFirstModifyTime = callbackResp.MsgFirstModifyTime
rResp.Result = callbackResp.ResultReactionExtensionList
ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, isHistory, false)
return &rResp, 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)
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
}
if !isExists {
if !req.IsReact {
log.Debug(req.OperationID, "redis handle firstly", req.String())
rResp.MsgFirstModifyTime = utils.GetCurrentTimestampByMill()
for k, v := range req.ReactionExtensionList {
err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
if err != nil {
setKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, k, v)
continue
}
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
}
setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
}
rResp.IsReact = true
_, err := db.DB.SetMessageReactionExpire(req.ClientMsgID, req.SessionType, time.Duration(24*3)*time.Hour)
if err != nil {
log.Error(req.OperationID, "SetMessageReactionExpire err:", err.Error(), req.String())
}
} else {
err := rpc.dMessageLocker.LockGlobalMessage(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
}
mongoValue, err := db.DB.GetExtendMsg(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
}
setValue := make(map[string]*server_api_params.KeyValue)
for k, v := range req.ReactionExtensionList {
temp := new(server_api_params.KeyValue)
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)
continue
}
}
temp.TypeKey = k
temp.Value = v.Value
temp.LatestUpdateTime = utils.GetCurrentTimestampByMill()
setValue[k] = temp
}
err = db.DB.InsertOrUpdateReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, setValue)
if err != nil {
for _, value := range setValue {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
} else {
for _, value := range setValue {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
rResp.Result = append(rResp.Result, temp)
}
}
lockErr := rpc.dMessageLocker.UnLockGlobalMessage(req.ClientMsgID)
if lockErr != nil {
log.Error(req.OperationID, "UnLockGlobalMessage err:", lockErr.Error())
}
}
} else {
log.Debug(req.OperationID, "redis handle secondly", req.String())
for k, v := range req.ReactionExtensionList {
err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
if err != nil {
setKeyResultInfo(&rResp, 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)
continue
}
temp := new(server_api_params.KeyValue)
utils.JsonStringToStruct(redisValue, temp)
if v.LatestUpdateTime != temp.LatestUpdateTime {
setKeyResultInfo(&rResp, 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)
continue
}
setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
}
}
}
if !isExists {
if !req.IsReact {
ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, true, true)
} else {
ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, false)
}
} else {
ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, true)
}
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
return &rResp, nil
}
func setKeyResultInfo(r *msg.SetMessageReactionExtensionsResp, errCode int32, errMsg, clientMsgID, typeKey string, keyValue *server_api_params.KeyValue) {
temp := new(msg.KeyValueResp)
temp.KeyValue = keyValue
temp.ErrCode = errCode
temp.ErrMsg = errMsg
r.Result = append(r.Result, temp)
_ = db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
}
func setDeleteKeyResultInfo(r *msg.DeleteMessageListReactionExtensionsResp, errCode int32, errMsg, clientMsgID, typeKey string, keyValue *server_api_params.KeyValue) {
temp := new(msg.KeyValueResp)
temp.KeyValue = keyValue
temp.ErrCode = errCode
temp.ErrMsg = errMsg
r.Result = append(r.Result, temp)
_ = db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
}
func (rpc *rpcChat) GetMessageListReactionExtensions(ctx context.Context, req *msg.GetMessageListReactionExtensionsReq) (resp *msg.GetMessageListReactionExtensionsResp, err error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
var rResp msg.GetMessageListReactionExtensionsResp
if req.IsExternalExtensions {
callbackResp := callbackGetMessageListReactionExtensions(req)
if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 {
rResp.ErrCode = int32(callbackResp.ErrCode)
rResp.ErrMsg = callbackResp.ErrMsg
return &rResp, nil
} else {
rResp.SingleMessageResult = callbackResp.MessageResultList
return &rResp, nil
}
}
for _, messageValue := range req.MessageReactionKeyList {
var oneMessage msg.SingleMessageExtensionResult
oneMessage.ClientMsgID = messageValue.ClientMsgID
isExists, err := db.DB.JudgeMessageReactionEXISTS(messageValue.ClientMsgID, req.SessionType)
if err != nil {
rResp.ErrCode = 100
rResp.ErrMsg = err.Error()
return &rResp, nil
}
if isExists {
redisValue, err := db.DB.GetOneMessageAllReactionList(messageValue.ClientMsgID, req.SessionType)
if err != nil {
oneMessage.ErrCode = 100
oneMessage.ErrMsg = err.Error()
rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
continue
}
keyMap := make(map[string]*server_api_params.KeyValue)
for k, v := range redisValue {
temp := new(server_api_params.KeyValue)
utils.JsonStringToStruct(v, temp)
keyMap[k] = temp
}
oneMessage.ReactionExtensionList = keyMap
} else {
mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, messageValue.ClientMsgID, messageValue.MsgFirstModifyTime)
if err != nil {
oneMessage.ErrCode = 100
oneMessage.ErrMsg = err.Error()
rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
continue
}
keyMap := make(map[string]*server_api_params.KeyValue)
for k, v := range mongoValue.ReactionExtensionList {
temp := new(server_api_params.KeyValue)
temp.TypeKey = v.TypeKey
temp.Value = v.Value
temp.LatestUpdateTime = v.LatestUpdateTime
keyMap[k] = temp
}
oneMessage.ReactionExtensionList = keyMap
}
rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
}
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
return &rResp, nil
}
func (rpc *rpcChat) AddMessageReactionExtensions(ctx context.Context, req *msg.AddMessageReactionExtensionsReq) (resp *msg.AddMessageReactionExtensionsResp, err error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
var rResp msg.AddMessageReactionExtensionsResp
rResp.ClientMsgID = req.ClientMsgID
rResp.MsgFirstModifyTime = req.MsgFirstModifyTime
callbackResp := callbackAddMessageReactionExtensions(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
}
//if !req.IsExternalExtensions {
// rResp.ErrCode = 200
// rResp.ErrMsg = "only extenalextensions message can be used"
// 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
//}
//if ExternalExtension
var isHistory bool
if req.IsReact {
isHistory = false
} else {
isHistory = true
}
rResp.MsgFirstModifyTime = callbackResp.MsgFirstModifyTime
rResp.Result = callbackResp.ResultReactionExtensionList
rResp.IsReact = callbackResp.IsReact
ExtendMessageAddedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, isHistory, false)
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", resp.String())
return &rResp, nil
}
func (rpc *rpcChat) DeleteMessageReactionExtensions(ctx context.Context, req *msg.DeleteMessageListReactionExtensionsReq) (resp *msg.DeleteMessageListReactionExtensionsResp, err error) {
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
var rResp msg.DeleteMessageListReactionExtensionsResp
callbackResp := callbackDeleteMessageReactionExtensions(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
}
//if ExternalExtension
if req.IsExternalExtensions {
rResp.Result = callbackResp.ResultReactionExtensionList
ExtendMessageDeleteNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, false)
return &rResp, nil
}
for _, v := range callbackResp.ResultReactionExtensionList {
if v.ErrCode != 0 {
func(req *[]*server_api_params.KeyValue, typeKey string) {
for i := 0; i < len(*req); i++ {
if (*req)[i].TypeKey == typeKey {
*req = append((*req)[:i], (*req)[i+1:]...)
}
}
}(&req.ReactionExtensionList, v.KeyValue.TypeKey)
rResp.Result = append(rResp.Result, v)
}
}
isExists, err := db.DB.JudgeMessageReactionEXISTS(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
}
if isExists {
log.Debug(req.OperationID, "redis handle this delete", req.String())
for _, v := range req.ReactionExtensionList {
err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, v.TypeKey)
if err != nil {
setDeleteKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, v.TypeKey, v)
continue
}
redisValue, err := db.DB.GetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, v.TypeKey)
if err != nil && err != go_redis.Nil {
setDeleteKeyResultInfo(&rResp, 200, err.Error(), req.ClientMsgID, v.TypeKey, v)
continue
}
temp := new(server_api_params.KeyValue)
utils.JsonStringToStruct(redisValue, temp)
if v.LatestUpdateTime != temp.LatestUpdateTime {
setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp)
continue
} else {
newErr := db.DB.DeleteOneMessageKey(req.ClientMsgID, req.SessionType, v.TypeKey)
if newErr != nil {
setDeleteKeyResultInfo(&rResp, 201, newErr.Error(), req.ClientMsgID, v.TypeKey, temp)
continue
}
setDeleteKeyResultInfo(&rResp, 0, "", req.ClientMsgID, v.TypeKey, v)
}
}
} else {
err := rpc.dMessageLocker.LockGlobalMessage(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
}
mongoValue, err := db.DB.GetExtendMsg(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
}
setValue := make(map[string]*server_api_params.KeyValue)
for _, v := range req.ReactionExtensionList {
temp := new(server_api_params.KeyValue)
if vv, ok := mongoValue.ReactionExtensionList[v.TypeKey]; ok {
utils.CopyStructFields(temp, &vv)
if v.LatestUpdateTime != vv.LatestUpdateTime {
setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp)
continue
}
} else {
setDeleteKeyResultInfo(&rResp, 400, "key not in", req.ClientMsgID, v.TypeKey, v)
continue
}
temp.TypeKey = v.TypeKey
setValue[v.TypeKey] = temp
}
err = db.DB.DeleteReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, setValue)
if err != nil {
for _, value := range setValue {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
temp.ErrMsg = err.Error()
temp.ErrCode = 100
rResp.Result = append(rResp.Result, temp)
}
} else {
for _, value := range setValue {
temp := new(msg.KeyValueResp)
temp.KeyValue = value
rResp.Result = append(rResp.Result, temp)
}
}
lockErr := rpc.dMessageLocker.UnLockGlobalMessage(req.ClientMsgID)
if lockErr != nil {
log.Error(req.OperationID, "UnLockGlobalMessage err:", lockErr.Error())
}
}
ExtendMessageDeleteNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, isExists)
log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
return &rResp, nil
}
internal/rpc/msg/extend_msg.notification.go
+131
View File
@@ -0,0 +1,131 @@
package msg
import (
"Open_IM/pkg/base_info"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
"Open_IM/pkg/grpc-etcdv3/getcdv3"
"Open_IM/pkg/proto/msg"
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"context"
"strings"
)
func ExtendMessageUpdatedNotification(operationID, sendID string, sourceID string, sessionType int32,
req *msg.SetMessageReactionExtensionsReq, resp *msg.SetMessageReactionExtensionsResp, isHistory bool, isReactionFromCache bool) {
var m base_info.ReactionMessageModifierNotification
m.SourceID = req.SourceID
m.OpUserID = req.OpUserID
m.Operation = constant.SetMessageExtensions
m.SessionType = req.SessionType
keyMap := make(map[string]*open_im_sdk.KeyValue)
for _, valueResp := range resp.Result {
if valueResp.ErrCode == 0 {
keyMap[valueResp.KeyValue.TypeKey] = valueResp.KeyValue
}
}
if len(keyMap) == 0 {
log.NewWarn(operationID, "all key set failed can not send notification", *req)
return
}
m.SuccessReactionExtensionList = keyMap
m.ClientMsgID = req.ClientMsgID
m.IsReact = resp.IsReact
m.IsExternalExtensions = req.IsExternalExtensions
m.MsgFirstModifyTime = resp.MsgFirstModifyTime
messageReactionSender(operationID, sendID, sourceID, sessionType, constant.ReactionMessageModifier, utils.StructToJsonString(m), isHistory, isReactionFromCache)
}
func ExtendMessageAddedNotification(operationID, sendID string, sourceID string, sessionType int32,
req *msg.AddMessageReactionExtensionsReq, resp *msg.AddMessageReactionExtensionsResp, isHistory bool, isReactionFromCache bool) {
var m base_info.ReactionMessageModifierNotification
m.SourceID = req.SourceID
m.OpUserID = req.OpUserID
m.Operation = constant.AddMessageExtensions
m.SessionType = req.SessionType
keyMap := make(map[string]*open_im_sdk.KeyValue)
for _, valueResp := range resp.Result {
if valueResp.ErrCode == 0 {
keyMap[valueResp.KeyValue.TypeKey] = valueResp.KeyValue
}
}
if len(keyMap) == 0 {
log.NewWarn(operationID, "all key set failed can not send notification", *req)
return
}
m.SuccessReactionExtensionList = keyMap
m.ClientMsgID = req.ClientMsgID
m.IsReact = resp.IsReact
m.IsExternalExtensions = req.IsExternalExtensions
m.MsgFirstModifyTime = resp.MsgFirstModifyTime
messageReactionSender(operationID, sendID, sourceID, sessionType, constant.ReactionMessageModifier, utils.StructToJsonString(m), isHistory, isReactionFromCache)
}
func ExtendMessageDeleteNotification(operationID, sendID string, sourceID string, sessionType int32,
req *msg.DeleteMessageListReactionExtensionsReq, resp *msg.DeleteMessageListReactionExtensionsResp, isHistory bool, isReactionFromCache bool) {
var m base_info.ReactionMessageDeleteNotification
m.SourceID = req.SourceID
m.OpUserID = req.OpUserID
m.SessionType = req.SessionType
keyMap := make(map[string]*open_im_sdk.KeyValue)
for _, valueResp := range resp.Result {
if valueResp.ErrCode == 0 {
keyMap[valueResp.KeyValue.TypeKey] = valueResp.KeyValue
}
}
if len(keyMap) == 0 {
log.NewWarn(operationID, "all key set failed can not send notification", *req)
return
}
m.SuccessReactionExtensionList = keyMap
m.ClientMsgID = req.ClientMsgID
m.MsgFirstModifyTime = req.MsgFirstModifyTime
messageReactionSender(operationID, sendID, sourceID, sessionType, constant.ReactionMessageDeleter, utils.StructToJsonString(m), isHistory, isReactionFromCache)
}
func messageReactionSender(operationID, sendID string, sourceID string, sessionType, contentType int32, content string, isHistory bool, isReactionFromCache bool) {
options := make(map[string]bool, 5)
utils.SetSwitchFromOptions(options, constant.IsOfflinePush, false)
utils.SetSwitchFromOptions(options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(options, constant.IsSenderConversationUpdate, false)
utils.SetSwitchFromOptions(options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(options, constant.IsReactionFromCache, isReactionFromCache)
if !isHistory {
utils.SetSwitchFromOptions(options, constant.IsHistory, false)
utils.SetSwitchFromOptions(options, constant.IsPersistent, false)
}
pbData := msg.SendMsgReq{
OperationID: operationID,
MsgData: &open_im_sdk.MsgData{
SendID: sendID,
ClientMsgID: utils.GetMsgID(sendID),
SessionType: sessionType,
MsgFrom: constant.SysMsgType,
ContentType: contentType,
Content: []byte(content),
// ForceList: params.ForceList,
CreateTime: utils.GetCurrentTimestampByMill(),
Options: options,
},
}
switch sessionType {
case constant.SingleChatType, constant.NotificationChatType:
pbData.MsgData.RecvID = sourceID
case constant.GroupChatType, constant.SuperGroupChatType:
pbData.MsgData.GroupID = sourceID
}
etcdConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, operationID)
if etcdConn == nil {
errMsg := operationID + "getcdv3.GetDefaultConn == nil"
log.NewError(operationID, errMsg)
return
}
client := msg.NewMsgClient(etcdConn)
reply, err := client.SendMsg(context.Background(), &pbData)
if err != nil {
log.NewError(operationID, "SendMsg rpc failed, ", pbData.String(), err.Error())
} else if reply.ErrCode != 0 {
log.NewError(operationID, "SendMsg rpc failed, ", pbData.String(), reply.ErrCode, reply.ErrMsg)
}
}
internal/rpc/msg/extend_msg_callback.go
+104
View File
@@ -0,0 +1,104 @@
package msg
import (
cbApi "Open_IM/pkg/call_back_struct"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/http"
"Open_IM/pkg/common/log"
"Open_IM/pkg/proto/msg"
"Open_IM/pkg/utils"
http2 "net/http"
)
func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensionsReq) *cbApi.CallbackBeforeSetMessageReactionExtResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
req := cbApi.CallbackBeforeSetMessageReactionExtReq{
OperationID: setReq.OperationID,
CallbackCommand: constant.CallbackBeforeSetMessageReactionExtensionCommand,
SourceID: setReq.SourceID,
OpUserID: setReq.OpUserID,
SessionType: setReq.SessionType,
ReactionExtensionList: setReq.ReactionExtensionList,
ClientMsgID: setReq.ClientMsgID,
IsReact: setReq.IsReact,
IsExternalExtensions: setReq.IsExternalExtensions,
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbApi.CallbackBeforeSetMessageReactionExtResp{CommonCallbackResp: &callbackResp}
defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp)
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetMessageReactionExtensionCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
callbackResp.ErrCode = http2.StatusInternalServerError
callbackResp.ErrMsg = err.Error()
}
return resp
}
func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReactionExtensionsReq) *cbApi.CallbackDeleteMessageReactionExtResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
req := cbApi.CallbackDeleteMessageReactionExtReq{
OperationID: setReq.OperationID,
CallbackCommand: constant.CallbackBeforeDeleteMessageReactionExtensionsCommand,
SourceID: setReq.SourceID,
OpUserID: setReq.OpUserID,
SessionType: setReq.SessionType,
ReactionExtensionList: setReq.ReactionExtensionList,
ClientMsgID: setReq.ClientMsgID,
IsExternalExtensions: setReq.IsExternalExtensions,
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbApi.CallbackDeleteMessageReactionExtResp{CommonCallbackResp: &callbackResp}
defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp)
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeDeleteMessageReactionExtensionsCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
callbackResp.ErrCode = http2.StatusInternalServerError
callbackResp.ErrMsg = err.Error()
}
return resp
}
func callbackGetMessageListReactionExtensions(getReq *msg.GetMessageListReactionExtensionsReq) *cbApi.CallbackGetMessageListReactionExtResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: getReq.OperationID}
log.NewDebug(getReq.OperationID, utils.GetSelfFuncName(), getReq.String())
req := cbApi.CallbackGetMessageListReactionExtReq{
OperationID: getReq.OperationID,
CallbackCommand: constant.CallbackGetMessageListReactionExtensionsCommand,
SourceID: getReq.SourceID,
OpUserID: getReq.OpUserID,
SessionType: getReq.SessionType,
TypeKeyList: getReq.TypeKeyList,
MessageKeyList: getReq.MessageReactionKeyList,
}
resp := &cbApi.CallbackGetMessageListReactionExtResp{CommonCallbackResp: &callbackResp}
defer log.NewDebug(getReq.OperationID, utils.GetSelfFuncName(), req, *resp)
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackGetMessageListReactionExtensionsCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
callbackResp.ErrCode = http2.StatusInternalServerError
callbackResp.ErrMsg = err.Error()
}
return resp
}
func callbackAddMessageReactionExtensions(setReq *msg.AddMessageReactionExtensionsReq) *cbApi.CallbackAddMessageReactionExtResp {
callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
req := cbApi.CallbackAddMessageReactionExtReq{
OperationID: setReq.OperationID,
CallbackCommand: constant.CallbackAddMessageListReactionExtensionsCommand,
SourceID: setReq.SourceID,
OpUserID: setReq.OpUserID,
SessionType: setReq.SessionType,
ReactionExtensionList: setReq.ReactionExtensionList,
ClientMsgID: setReq.ClientMsgID,
IsReact: setReq.IsReact,
IsExternalExtensions: setReq.IsExternalExtensions,
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbApi.CallbackAddMessageReactionExtResp{CommonCallbackResp: &callbackResp}
defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp, *resp.CommonCallbackResp, resp.IsReact, resp.MsgFirstModifyTime)
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackAddMessageListReactionExtensionsCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
callbackResp.ErrCode = http2.StatusInternalServerError
callbackResp.ErrMsg = err.Error()
}
return resp
}
internal/rpc/msg/group_notification.go
+35 -1
View File
@@ -245,12 +245,15 @@ func GroupCreatedNotification(operationID, opUserID, groupID string, initMemberL
// notification := ""
// introduction := ""
// faceURL := ""
func GroupInfoSetNotification(operationID, opUserID, groupID string, groupName, notification, introduction, faceURL string, needVerification *wrapperspb.Int32Value) {
func GroupInfoSetNotification(operationID, opUserID, groupID string, groupName, notification, introduction, faceURL string, needVerification, applyMemberFriend, lookMemberInfo *wrapperspb.Int32Value) {
GroupInfoChangedTips := open_im_sdk.GroupInfoSetTips{Group: &open_im_sdk.GroupInfo{}, OpUser: &open_im_sdk.GroupMemberFullInfo{}}
if err := setGroupInfo(groupID, GroupInfoChangedTips.Group); err != nil {
log.Error(operationID, "setGroupInfo failed ", err.Error(), groupID)
return
}
GroupInfoChangedTips.Group.NeedVerification = 0
GroupInfoChangedTips.Group.LookMemberInfo = 0
GroupInfoChangedTips.Group.ApplyMemberFriend = 0
GroupInfoChangedTips.Group.GroupName = groupName
GroupInfoChangedTips.Group.Notification = notification
GroupInfoChangedTips.Group.Introduction = introduction
@@ -258,6 +261,12 @@ func GroupInfoSetNotification(operationID, opUserID, groupID string, groupName,
if needVerification != nil {
GroupInfoChangedTips.Group.NeedVerification = needVerification.Value
}
if applyMemberFriend != nil {
GroupInfoChangedTips.Group.ApplyMemberFriend = applyMemberFriend.Value
}
if lookMemberInfo != nil {
GroupInfoChangedTips.Group.LookMemberInfo = lookMemberInfo.Value
}
if err := setOpUserInfo(opUserID, groupID, GroupInfoChangedTips.OpUser); err != nil {
log.Error(operationID, "setOpUserInfo failed ", err.Error(), opUserID, groupID)
@@ -435,7 +444,20 @@ func GroupApplicationAcceptedNotification(req *pbGroup.GroupApplicationResponseR
log.Error(req.OperationID, "setOpUserInfo failed", req.OpUserID, req.GroupID, GroupApplicationAcceptedTips.OpUser)
return
}
groupNotification(constant.GroupApplicationAcceptedNotification, &GroupApplicationAcceptedTips, req.OpUserID, "", req.FromUserID, req.OperationID)
adminList, err := imdb.GetOwnerManagerByGroupID(req.GroupID)
if err != nil {
log.Error(req.OperationID, "GetOwnerManagerByGroupID failed", req.GroupID)
return
}
for _, v := range adminList {
if v.UserID == req.OpUserID {
continue
}
GroupApplicationAcceptedTips.ReceiverAs = 1
groupNotification(constant.GroupApplicationAcceptedNotification, &GroupApplicationAcceptedTips, req.OpUserID, "", v.UserID, req.OperationID)
}
}
func GroupApplicationRejectedNotification(req *pbGroup.GroupApplicationResponseReq) {
@@ -449,6 +471,18 @@ func GroupApplicationRejectedNotification(req *pbGroup.GroupApplicationResponseR
return
}
groupNotification(constant.GroupApplicationRejectedNotification, &GroupApplicationRejectedTips, req.OpUserID, "", req.FromUserID, req.OperationID)
adminList, err := imdb.GetOwnerManagerByGroupID(req.GroupID)
if err != nil {
log.Error(req.OperationID, "GetOwnerManagerByGroupID failed", req.GroupID)
return
}
for _, v := range adminList {
if v.UserID == req.OpUserID {
continue
}
GroupApplicationRejectedTips.ReceiverAs = 1
groupNotification(constant.GroupApplicationRejectedNotification, &GroupApplicationRejectedTips, req.OpUserID, "", v.UserID, req.OperationID)
}
}
func GroupOwnerTransferredNotification(req *pbGroup.TransferGroupOwnerReq) {
internal/rpc/msg/lock.go
+52
View File
@@ -0,0 +1,52 @@
package msg
import (
"Open_IM/pkg/common/db"
"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)
}
type LockerMessage struct{}
func NewLockerMessage() *LockerMessage {
return &LockerMessage{}
}
func (l *LockerMessage) LockMessageTypeKey(clientMsgID, typeKey string) (err error) {
for i := 0; i < 3; i++ {
err = db.DB.LockMessageTypeKey(clientMsgID, typeKey)
if err != nil {
time.Sleep(time.Millisecond * 100)
continue
} else {
break
}
}
return err
}
func (l *LockerMessage) LockGlobalMessage(clientMsgID string) (err error) {
for i := 0; i < 3; i++ {
err = db.DB.LockMessageTypeKey(clientMsgID, GlOBLLOCK)
if err != nil {
time.Sleep(time.Millisecond * 100)
continue
} else {
break
}
}
return err
}
func (l *LockerMessage) UnLockMessageTypeKey(clientMsgID string, typeKey string) error {
return db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
}
func (l *LockerMessage) UnLockGlobalMessage(clientMsgID string) error {
return db.DB.UnLockMessageTypeKey(clientMsgID, GlOBLLOCK)
}
internal/rpc/msg/rpcChat.go
+6 -9
View File
@@ -31,7 +31,8 @@ type rpcChat struct {
etcdAddr []string
messageWriter MessageWriter
//offlineProducer *kafka.Producer
delMsgCh chan deleteMsg
delMsgCh chan deleteMsg
dMessageLocker MessageLocker
}
type deleteMsg struct {
@@ -48,6 +49,7 @@ func NewRpcChatServer(port int) *rpcChat {
rpcRegisterName: config.Config.RpcRegisterName.OpenImMsgName,
etcdSchema: config.Config.Etcd.EtcdSchema,
etcdAddr: config.Config.Etcd.EtcdAddr,
dMessageLocker: NewLockerMessage(),
}
rc.messageWriter = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
//rc.offlineProducer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschatOffline.Addr, config.Config.Kafka.Ws2mschatOffline.Topic)
@@ -143,14 +145,9 @@ func (rpc *rpcChat) runCh() {
select {
case msg := <-rpc.delMsgCh:
log.NewInfo(msg.OperationID, utils.GetSelfFuncName(), "delmsgch recv new: ", msg)
db.DB.DelMsgFromCache(msg.UserID, msg.SeqList, msg.OperationID)
unexistSeqList, err := db.DB.DelMsgBySeqList(msg.UserID, msg.SeqList, msg.OperationID)
if err != nil {
log.NewError(msg.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", msg.UserID, msg.SeqList, msg.OperationID, err.Error())
continue
}
if len(unexistSeqList) > 0 {
DeleteMessageNotification(msg.OpUserID, msg.UserID, unexistSeqList, msg.OperationID)
if len(msg.SeqList) > 0 {
db.DB.DelMsgFromCache(msg.UserID, msg.SeqList, msg.OperationID)
DeleteMessageNotification(msg.OpUserID, msg.UserID, msg.SeqList, msg.OperationID)
}
}
}
internal/rpc/msg/send_msg.go
+2 -1
View File
@@ -25,6 +25,7 @@ import (
"time"
promePkg "Open_IM/pkg/common/prometheus"
go_redis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
)
@@ -125,7 +126,7 @@ func (rpc *rpcChat) messageVerification(data *pbChat.SendMsgReq) (bool, int32, s
if utils.IsContain(data.MsgData.SendID, config.Config.Manager.AppManagerUid) {
return true, 0, "", nil
}
if data.MsgData.ContentType <= constant.NotificationEnd && data.MsgData.ContentType >= constant.NotificationBegin {
if data.MsgData.ContentType <= constant.NotificationEnd && data.MsgData.ContentType >= constant.NotificationBegin && data.MsgData.ContentType != constant.SignalingNotification {
return true, 0, "", nil
}
log.NewDebug(data.OperationID, *config.Config.MessageVerify.FriendVerify)
internal/rpc/user/user.go
+50 -1
View File
@@ -163,6 +163,22 @@ func (s *userServer) BatchSetConversations(ctx context.Context, req *pbUser.Batc
return resp, nil
}
if v.ConversationType == constant.SuperGroupChatType {
if v.RecvMsgOpt == constant.ReceiveNotNotifyMessage {
if err := db.DB.SetSuperGroupUserReceiveNotNotifyMessage(v.GroupID, v.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), v.GroupID, v.OwnerUserID)
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: err.Error()}
return resp, nil
}
} else {
if err := db.DB.SetSuperGroupUserReceiveNotifyMessage(v.GroupID, v.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), v.GroupID, err.Error())
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: err.Error()}
return resp, nil
}
}
}
isUpdate, err := imdb.SetConversation(conversation)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "SetConversation error", err.Error())
@@ -251,7 +267,7 @@ func (s *userServer) SetConversation(ctx context.Context, req *pbUser.SetConvers
if req.NotificationType == 0 {
req.NotificationType = constant.ConversationOptChangeNotification
}
if req.Conversation.ConversationType == constant.GroupChatType {
if req.Conversation.ConversationType == constant.GroupChatType || req.Conversation.ConversationType == constant.SuperGroupChatType {
groupInfo, err := imdb.GetGroupInfoByGroupID(req.Conversation.GroupID)
if err != nil {
log.NewError(req.OperationID, "GetGroupInfoByGroupID failed ", req.Conversation.GroupID, err.Error())
@@ -264,7 +280,24 @@ func (s *userServer) SetConversation(ctx context.Context, req *pbUser.SetConvers
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: errMsg}
return resp, nil
}
if req.Conversation.ConversationType == constant.SuperGroupChatType {
if req.Conversation.RecvMsgOpt == constant.ReceiveNotNotifyMessage {
if err = db.DB.SetSuperGroupUserReceiveNotNotifyMessage(req.Conversation.GroupID, req.Conversation.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), req.Conversation.GroupID, req.Conversation.OwnerUserID)
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: err.Error()}
return resp, nil
}
} else {
if err = db.DB.SetSuperGroupUserReceiveNotifyMessage(req.Conversation.GroupID, req.Conversation.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), req.Conversation.GroupID, err.Error())
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: err.Error()}
return resp, nil
}
}
}
}
var conversation db.Conversation
if err := utils.CopyStructFields(&conversation, req.Conversation); err != nil {
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields failed", *req.Conversation, err.Error())
@@ -326,6 +359,21 @@ func (s *userServer) SetRecvMsgOpt(ctx context.Context, req *pbUser.SetRecvMsgOp
case "group":
conversation.GroupID = stringList[1]
conversation.ConversationType = constant.GroupChatType
case "super_group":
conversation.GroupID = stringList[1]
if req.RecvMsgOpt == constant.ReceiveNotNotifyMessage {
if err := db.DB.SetSuperGroupUserReceiveNotNotifyMessage(conversation.GroupID, req.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), conversation.GroupID, req.OwnerUserID)
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
} else {
if err := db.DB.SetSuperGroupUserReceiveNotifyMessage(conversation.GroupID, req.OwnerUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error(), conversation.GroupID, req.OwnerUserID)
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
}
}
}
isUpdate, err := imdb.SetRecvMsgOpt(conversation)
@@ -334,6 +382,7 @@ func (s *userServer) SetRecvMsgOpt(ctx context.Context, req *pbUser.SetRecvMsgOp
resp.CommonResp = &pbUser.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
return resp, nil
}
if isUpdate {
err = rocksCache.DelConversationFromCache(conversation.OwnerUserID, conversation.ConversationID)
} else {