msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/tuoyun' into tuoyun

Browse Source
pull/218/head
wenxu12345 3 years ago
parent 2c1bd41fba 6260cf24cc
commit 57a746fc21
16 changed files with 1188 additions and 502 deletions
Show Stats Download Patch File Download Diff File

2
cmd/open_im_api/main.go
Unescape View File

@ -43,6 +43,8 @@ func main() {
userRouterGroup.POST("/get_self_user_info", user.GetSelfUserInfo) //1 userRouterGroup.POST("/get_self_user_info", user.GetSelfUserInfo) //1
userRouterGroup.POST("/get_users_online_status", user.GetUsersOnlineStatus) //1 userRouterGroup.POST("/get_users_online_status", user.GetUsersOnlineStatus) //1
userRouterGroup.POST("/get_users_info_from_cache", user.GetUsersInfoFromCache) userRouterGroup.POST("/get_users_info_from_cache", user.GetUsersInfoFromCache)
userRouterGroup.POST("/get_user_friend_from_cache", user.GetFriendIDListFromCache)
userRouterGroup.POST("/get_black_list_from_cache", user.GetBlackIDListFromCache)
} }
//friend routing group //friend routing group
friendRouterGroup := r.Group("/friend") friendRouterGroup := r.Group("/friend")

4
internal/api/office/work_moments.go
Unescape View File

@ -267,7 +267,7 @@ func GetUserWorkMoments(c *gin.Context) {
UserID: comment.UserID, UserID: comment.UserID,
UserName: comment.UserName, UserName: comment.UserName,
ReplyUserID: comment.ReplyUserID, ReplyUserID: comment.ReplyUserID,
ReplyUserName: comment.ReplyUserID, ReplyUserName: comment.ReplyUserName,
ContentID: comment.ContentID, ContentID: comment.ContentID,
Content: comment.Content, Content: comment.Content,
CreateTime: comment.CreateTime, CreateTime: comment.CreateTime,
@ -350,7 +350,7 @@ func GetUserFriendWorkMoments(c *gin.Context) {
UserID: comment.UserID, UserID: comment.UserID,
UserName: comment.UserName, UserName: comment.UserName,
ReplyUserID: comment.ReplyUserID, ReplyUserID: comment.ReplyUserID,
ReplyUserName: comment.ReplyUserID, ReplyUserName: comment.ReplyUserName,
ContentID: comment.ContentID, ContentID: comment.ContentID,
Content: comment.Content, Content: comment.Content,
CreateTime: comment.CreateTime, CreateTime: comment.CreateTime,

125
internal/api/user/user.go
Unescape View File

@ -8,6 +8,7 @@ import (
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/common/token_verify" "Open_IM/pkg/common/token_verify"
"Open_IM/pkg/grpc-etcdv3/getcdv3" "Open_IM/pkg/grpc-etcdv3/getcdv3"
cacheRpc "Open_IM/pkg/proto/cache"
pbRelay "Open_IM/pkg/proto/relay" pbRelay "Open_IM/pkg/proto/relay"
open_im_sdk "Open_IM/pkg/proto/sdk_ws" open_im_sdk "Open_IM/pkg/proto/sdk_ws"
rpc "Open_IM/pkg/proto/user" rpc "Open_IM/pkg/proto/user"
@ -18,6 +19,44 @@ import (
"strings" "strings"
) )
func GetUsersInfoFromCache(c *gin.Context) {
params := api.GetUsersInfoReq{}
if err := c.BindJSON(&params); err != nil {
log.NewError("0", "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": http.StatusBadRequest, "errMsg": err.Error()})
return
}
req := &rpc.GetUserInfoReq{}
utils.CopyStructFields(req, &params)
var ok bool
ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
return
}
log.NewInfo(params.OperationID, "GetUserInfo args ", req.String())
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
client := rpc.NewUserClient(etcdConn)
RpcResp, err := client.GetUserInfo(context.Background(), req)
if err != nil {
log.NewError(req.OperationID, "GetUserInfo failed ", err.Error(), req.String())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed"})
return
}
var publicUserInfoList []*open_im_sdk.PublicUserInfo
for _, v := range RpcResp.UserInfoList {
publicUserInfoList = append(publicUserInfoList,
&open_im_sdk.PublicUserInfo{UserID: v.UserID, Nickname: v.Nickname, FaceURL: v.FaceURL, Gender: v.Gender, Ex: v.Ex})
}
resp := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList}
resp.Data = jsonData.JsonDataList(resp.UserInfoList)
log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp)
}
//func GetUsersInfoFromCache(c *gin.Context) { //func GetUsersInfoFromCache(c *gin.Context) {
// params := api.GetUsersInfoReq{} // params := api.GetUsersInfoReq{}
// if err := c.BindJSON(&params); err != nil { // if err := c.BindJSON(&params); err != nil {
@ -35,13 +74,13 @@ import (
// return // return
// } // }
// log.NewInfo(params.OperationID, "GetUserInfo args ", getUserInfoReq.String()) // log.NewInfo(params.OperationID, "GetUserInfo args ", getUserInfoReq.String())
// reqCacheGetUserInfo := &cacheRpc.GetUserInfoReq{} // reqCacheGetUserInfo := &cacheRpc.GetUserInfoFromCacheReq{}
// utils.CopyStructFields(reqCacheGetUserInfo, &params) // utils.CopyStructFields(reqCacheGetUserInfo, &params)
// var userInfoList []*open_im_sdk.UserInfo // var userInfoList []*open_im_sdk.UserInfo
// var publicUserInfoList []*open_im_sdk.PublicUserInfo // var publicUserInfoList []*open_im_sdk.PublicUserInfo
// etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName) // etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
// cacheClient := cacheRpc.NewCacheClient(etcdConn) // cacheClient := cacheRpc.NewCacheClient(etcdConn)
// cacheResp, err := cacheClient.GetUserInfo(context.Background(), reqCacheGetUserInfo) // cacheResp, err := cacheClient.GetUserInfoFromCache(context.Background(), reqCacheGetUserInfo)
// if err != nil { // if err != nil {
// log.NewError(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "GetUserInfo failed", err.Error()) // log.NewError(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "GetUserInfo failed", err.Error())
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed: " + err.Error()}) // c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed: " + err.Error()})
@ -101,11 +140,11 @@ import (
// return // return
// } // }
// userInfoList = append(userInfoList, rpcResp.UserInfoList...) // userInfoList = append(userInfoList, rpcResp.UserInfoList...)
// cacheUpdateUserInfoReq := &cacheRpc.UpdateUserInfoReq{ // cacheUpdateUserInfoReq := &cacheRpc.UpdateUserInfoToCacheReq{
// UserInfoList: rpcResp.UserInfoList, // UserInfoList: rpcResp.UserInfoList,
// OperationID: getUserInfoReq.OperationID, // OperationID: getUserInfoReq.OperationID,
// } // }
// _, err = cacheClient.UpdateUserInfo(context.Background(), cacheUpdateUserInfoReq) // _, err = cacheClient.UpdateUserInfoToCache(context.Background(), cacheUpdateUserInfoReq)
// if err != nil { // if err != nil {
// log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", err.Error()) // log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", err.Error())
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed:" + err.Error()}) // c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed:" + err.Error()})
@ -122,50 +161,70 @@ import (
// c.JSON(http.StatusOK, resp) // c.JSON(http.StatusOK, resp)
//} //}
func GetUsersInfoFromCache(c *gin.Context) { func GetFriendIDListFromCache(c *gin.Context) {
params := api.GetUsersInfoReq{} var (
if err := c.BindJSON(&params); err != nil { req api.GetFriendIDListFromCacheReq
log.NewError("0", "BindJSON failed ", err.Error()) resp api.GetFriendIDListFromCacheResp
reqPb cacheRpc.GetFriendIDListFromCacheReq
respPb *cacheRpc.GetFriendIDListFromCacheResp
)
if err := c.BindJSON(&req); err != nil {
log.NewError(req.OperationID, "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": http.StatusBadRequest, "errMsg": err.Error()}) c.JSON(http.StatusBadRequest, gin.H{"errCode": http.StatusBadRequest, "errMsg": err.Error()})
return return
} }
req := &rpc.GetUserInfoReq{} reqPb.OperationID = req.OperationID
utils.CopyStructFields(req, &params)
var ok bool var ok bool
ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID) ok, reqPb.UserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok { if !ok {
log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token")) log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"}) c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
return return
} }
log.NewInfo(params.OperationID, "GetUserInfo args ", req.String()) etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
client := cacheRpc.NewCacheClient(etcdConn)
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName) respPb, err := client.GetFriendIDListFromCache(context.Background(), &reqPb)
client := rpc.NewUserClient(etcdConn)
RpcResp, err := client.GetUserInfo(context.Background(), req)
if err != nil { if err != nil {
log.NewError(req.OperationID, "GetUserInfo failed ", err.Error(), req.String()) log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetFriendIDListFromCache", err.Error())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed"}) c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed:" + err.Error()})
return return
} }
var publicUserInfoList []*open_im_sdk.PublicUserInfo resp.UserIDList = respPb.UserIDList
for _, v := range RpcResp.UserInfoList { resp.CommResp = api.CommResp{ErrMsg: respPb.CommonResp.ErrMsg, ErrCode: respPb.CommonResp.ErrCode}
publicUserInfoList = append(publicUserInfoList,
&open_im_sdk.PublicUserInfo{UserID: v.UserID, Nickname: v.Nickname, FaceURL: v.FaceURL, Gender: v.Gender, Ex: v.Ex})
}
resp := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList}
resp.Data = jsonData.JsonDataList(resp.UserInfoList)
log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp) c.JSON(http.StatusOK, resp)
} }
func GetUserFriendFromCache(c *gin.Context) { func GetBlackIDListFromCache(c *gin.Context) {
var (
} req api.GetBlackIDListFromCacheReq
resp api.GetBlackIDListFromCacheResp
func GetBlackListFromCache(c *gin.Context) { reqPb cacheRpc.GetBlackIDListFromCacheReq
respPb *cacheRpc.GetBlackIDListFromCacheResp
)
if err := c.BindJSON(&req); err != nil {
log.NewError(req.OperationID, "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": http.StatusBadRequest, "errMsg": err.Error()})
return
}
reqPb.OperationID = req.OperationID
var ok bool
ok, reqPb.UserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
return
}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
client := cacheRpc.NewCacheClient(etcdConn)
respPb, err := client.GetBlackIDListFromCache(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetFriendIDListFromCache", err.Error())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed:" + err.Error()})
return
}
resp.UserIDList = respPb.UserIDList
resp.CommResp = api.CommResp{ErrMsg: respPb.CommonResp.ErrMsg, ErrCode: respPb.CommonResp.ErrCode}
c.JSON(http.StatusOK, resp)
} }
func GetUsersInfo(c *gin.Context) { func GetUsersInfo(c *gin.Context) {

11
internal/msg_transfer/logic/history_msg_handler.go
Unescape View File

@ -14,6 +14,7 @@ import (
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"strings" "strings"
"time"
) )
type fcb func(msg []byte, msgKey string) type fcb func(msg []byte, msgKey string)
@ -38,7 +39,7 @@ func (mc *HistoryConsumerHandler) Init() {
} }
func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) { func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) {
time := utils.GetCurrentTimestampByNano() now := time.Now()
msgFromMQ := pbMsg.MsgDataToMQ{} msgFromMQ := pbMsg.MsgDataToMQ{}
err := proto.Unmarshal(msg, &msgFromMQ) err := proto.Unmarshal(msg, &msgFromMQ)
if err != nil { if err != nil {
@ -62,13 +63,13 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string)
return return
} }
mc.singleMsgCount++ mc.singleMsgCount++
log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", utils.GetCurrentTimestampByNano()-time) log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
} }
if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID { if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
} else { } else {
go sendMessageToPush(&msgFromMQ, msgKey) go sendMessageToPush(&msgFromMQ, msgKey)
} }
log.NewDebug(operationID, "saveUserChat cost time ", utils.GetCurrentTimestampByNano()-time) log.NewDebug(operationID, "saveUserChat cost time ", time.Since(now))
case constant.GroupChatType: case constant.GroupChatType:
log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = GroupChatType", isHistory, isPersist) log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = GroupChatType", isHistory, isPersist)
if isHistory { if isHistory {
@ -89,13 +90,13 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string)
return return
} }
mc.singleMsgCount++ mc.singleMsgCount++
log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", utils.GetCurrentTimestampByNano()-time) log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
} }
if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID { if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
} else { } else {
go sendMessageToPush(&msgFromMQ, msgKey) go sendMessageToPush(&msgFromMQ, msgKey)
} }
log.NewDebug(operationID, "saveUserChat cost time ", utils.GetCurrentTimestampByNano()-time) log.NewDebug(operationID, "saveUserChat cost time ", time.Since(now))
default: default:
log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String()) log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
return return

6
internal/push/logic/push_to_client.go
Unescape View File

@ -37,7 +37,7 @@ type AtContent struct {
func MsgToUser(pushMsg *pbPush.PushMsgReq) { func MsgToUser(pushMsg *pbPush.PushMsgReq) {
var wsResult []*pbRelay.SingleMsgToUser var wsResult []*pbRelay.SingleMsgToUser
isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush) isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
log.Debug("Get msg from msg_transfer And push msg", pushMsg.OperationID, "PushData", pushMsg.String()) log.Debug(pushMsg.OperationID, "Get msg from msg_transfer And push msg", pushMsg.String())
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName) grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
//Online push message //Online push message
log.Debug("test", pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String()) log.Debug("test", pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String())
@ -45,14 +45,14 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v) msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v)
reply, err := msgClient.OnlinePushMsg(context.Background(), &pbRelay.OnlinePushMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData, PushToUserID: pushMsg.PushToUserID}) reply, err := msgClient.OnlinePushMsg(context.Background(), &pbRelay.OnlinePushMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData, PushToUserID: pushMsg.PushToUserID})
if err != nil { if err != nil {
log.InfoByKv("push data to client rpc err", pushMsg.OperationID, "err", err) log.NewError("push data to client rpc err", pushMsg.OperationID, "err", err)
continue continue
} }
if reply != nil && reply.Resp != nil { if reply != nil && reply.Resp != nil {
wsResult = append(wsResult, reply.Resp...) wsResult = append(wsResult, reply.Resp...)
} }
} }
log.InfoByKv("push_result", pushMsg.OperationID, "result", wsResult, "sendData", pushMsg.MsgData) log.NewInfo(pushMsg.OperationID, "push_result", wsResult, "sendData", pushMsg.MsgData)
count++ count++
if isOfflinePush && pushMsg.PushToUserID != pushMsg.MsgData.SendID { if isOfflinePush && pushMsg.PushToUserID != pushMsg.MsgData.SendID {
for _, v := range wsResult { for _, v := range wsResult {

178
internal/rpc/cache/cache.go vendored
Unescape View File

@ -4,6 +4,7 @@ import (
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db" "Open_IM/pkg/common/db"
imdb "Open_IM/pkg/common/db/mysql_model/im_mysql_model"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/grpc-etcdv3/getcdv3" "Open_IM/pkg/grpc-etcdv3/getcdv3"
pbCache "Open_IM/pkg/proto/cache" pbCache "Open_IM/pkg/proto/cache"
@ -46,6 +47,14 @@ func (s *cacheServer) Run() {
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", registerAddress, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
//to cache
err = SyncDB2Cache()
if err != nil {
log.NewError("", err.Error(), "db to cache failed")
panic(err.Error())
}
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
pbCache.RegisterCacheServer(srv, s) pbCache.RegisterCacheServer(srv, s)
@ -62,14 +71,84 @@ func (s *cacheServer) Run() {
log.NewInfo("0", "message cms rpc success") log.NewInfo("0", "message cms rpc success")
} }
func (s *cacheServer) GetUserInfo(_ context.Context, req *pbCache.GetUserInfoReq) (resp *pbCache.GetUserInfoResp, err error) { func SyncDB2Cache() error {
var err error
log.NewInfo("0", utils.GetSelfFuncName())
userList, err := imdb.GetAllUser()
if err != nil {
return utils.Wrap(err, "")
}
//err = updateAllUserToCache(userList)
err = updateAllFriendToCache(userList)
err = updateAllBlackListToCache(userList)
return err
}
func updateAllUserToCache(userList []db.User) error {
for _, userInfo := range userList {
userInfoPb := &commonPb.UserInfo{
UserID: userInfo.UserID,
Nickname: userInfo.Nickname,
FaceURL: userInfo.FaceURL,
Gender: userInfo.Gender,
PhoneNumber: userInfo.PhoneNumber,
Birth: uint32(userInfo.Birth.Unix()),
Email: userInfo.Email,
Ex: userInfo.Ex,
CreateTime: uint32(userInfo.CreateTime.Unix()),
AppMangerLevel: userInfo.AppMangerLevel,
}
m, err := utils.Pb2Map(userInfoPb)
if err != nil {
log.NewError("", utils.GetSelfFuncName(), err.Error())
}
if err := db.DB.SetUserInfoToCache(userInfo.UserID, m); err != nil {
log.NewError("0", utils.GetSelfFuncName(), "set userInfo to cache failed", err.Error())
}
}
log.NewInfo("0", utils.GetSelfFuncName(), "ok")
return nil
}
func updateAllFriendToCache(userList []db.User) error {
log.NewInfo("0", utils.GetSelfFuncName())
for _, user := range userList {
friendIDList, err := imdb.GetFriendIDListByUserID(user.UserID)
if err != nil {
log.NewError("0", utils.GetSelfFuncName(), err.Error())
continue
}
if err := db.DB.AddFriendToCache(user.UserID, friendIDList...); err != nil {
log.NewError("0", utils.GetSelfFuncName(), err.Error())
}
}
log.NewInfo("0", utils.GetSelfFuncName(), "ok")
return nil
}
func updateAllBlackListToCache(userList []db.User) error {
log.NewInfo("0", utils.GetSelfFuncName())
for _, user := range userList {
blackIDList, err := imdb.GetBlackIDListByUserID(user.UserID)
if err != nil {
log.NewError("", utils.GetSelfFuncName(), err.Error())
continue
}
if err := db.DB.AddBlackUserToCache(user.UserID, blackIDList...); err != nil {
log.NewError("0", utils.GetSelfFuncName(), err.Error())
}
}
log.NewInfo("0", utils.GetSelfFuncName(), "ok")
return nil
}
func (s *cacheServer) GetUserInfoFromCache(_ context.Context, req *pbCache.GetUserInfoFromCacheReq) (resp *pbCache.GetUserInfoFromCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.GetUserInfoResp{ resp = &pbCache.GetUserInfoFromCacheResp{
UserInfoList: []*commonPb.UserInfo{}, CommonResp: &pbCache.CommonResp{},
CommonResp: &pbCache.CommonResp{},
} }
for _, userID := range req.UserIDList { for _, userID := range req.UserIDList {
userInfo, err := db.DB.GetUserInfo(userID) userInfo, err := db.DB.GetUserInfoFromCache(userID)
if err != nil { if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "get userInfo from cache failed", err.Error()) log.NewError(req.OperationID, utils.GetSelfFuncName(), "get userInfo from cache failed", err.Error())
continue continue
@ -80,39 +159,104 @@ func (s *cacheServer) GetUserInfo(_ context.Context, req *pbCache.GetUserInfoReq
return resp, nil return resp, nil
} }
func (s *cacheServer) UpdateUserInfo(_ context.Context, req *pbCache.UpdateUserInfoReq) (resp *pbCache.UpdateUserInfoResp, err error) { func (s *cacheServer) UpdateUserInfoToCache(_ context.Context, req *pbCache.UpdateUserInfoToCacheReq) (resp *pbCache.UpdateUserInfoToCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.UpdateUserInfoResp{ resp = &pbCache.UpdateUserInfoToCacheResp{
CommonResp: &pbCache.CommonResp{}, CommonResp: &pbCache.CommonResp{},
} }
for _, userInfo := range req.UserInfoList { for _, userInfo := range req.UserInfoList {
if err := db.DB.SetUserInfo(userInfo); err != nil { m, err := utils.Pb2Map(userInfo)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), *userInfo)
}
if err := db.DB.SetUserInfoToCache(userInfo.UserID, m); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "set userInfo to cache failed", err.Error()) log.NewError(req.OperationID, utils.GetSelfFuncName(), "set userInfo to cache failed", err.Error())
return resp, nil
} }
} }
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil return resp, nil
} }
func (s *cacheServer) GetBlackList(_ context.Context, req *pbCache.GetBlackListReq) (resp *pbCache.GetBlackListResp, err error) { func (s *cacheServer) GetFriendIDListFromCache(_ context.Context, req *pbCache.GetFriendIDListFromCacheReq) (resp *pbCache.GetFriendIDListFromCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.GetFriendIDListFromCacheResp{CommonResp: &pbCache.CommonResp{}}
friendIDList, err := db.DB.GetFriendIDListFromCache(req.UserID)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetFriendIDListFromCache", err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), friendIDList)
resp.UserIDList = friendIDList
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
}
func (s *cacheServer) AddFriendToCache(_ context.Context, req *pbCache.AddFriendToCacheReq) (resp *pbCache.AddFriendToCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
return nil, nil resp = &pbCache.AddFriendToCacheResp{CommonResp: &pbCache.CommonResp{}}
if err := db.DB.AddFriendToCache(req.UserID, req.FriendID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
} }
func (s *cacheServer) UpdateBlackList(_ context.Context, req *pbCache.UpdateBlackListReq) (resp *pbCache.UpdateBlackListResp, err error) { func (s *cacheServer) ReduceFriendFromCache(_ context.Context, req *pbCache.ReduceFriendFromCacheReq) (resp *pbCache.ReduceFriendFromCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.ReduceFriendFromCacheResp{CommonResp: &pbCache.CommonResp{}}
if err := db.DB.ReduceFriendToCache(req.UserID, req.FriendID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
}
func (s *cacheServer) GetBlackIDListFromCache(_ context.Context, req *pbCache.GetBlackIDListFromCacheReq) (resp *pbCache.GetBlackIDListFromCacheResp, err error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String()) log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
return nil, nil resp = &pbCache.GetBlackIDListFromCacheResp{CommonResp: &pbCache.CommonResp{}}
blackUserIDList, err := db.DB.GetBlackListFromCache(req.UserID)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), blackUserIDList)
resp.UserIDList = blackUserIDList
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
} }
func (s *cacheServer) GetFriendInfo(_ context.Context, req *pbCache.GetFriendInfoReq) (resp *pbCache.GetFriendInfoResp, err error) { func (s *cacheServer) AddBlackUserToCache(_ context.Context, req *pbCache.AddBlackUserToCacheReq) (resp *pbCache.AddBlackUserToCacheResp, err error) {
return nil, nil log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.AddBlackUserToCacheResp{CommonResp: &pbCache.CommonResp{}}
if err := db.DB.AddBlackUserToCache(req.UserID, req.BlackUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
} }
func (s *cacheServer) UpdateFriendInfo(_ context.Context, req *pbCache.UpdateFriendInfoReq) (resp *pbCache.UpdateFriendInfoResp, err error) { func (s *cacheServer) ReduceBlackUserFromCache(_ context.Context, req *pbCache.ReduceBlackUserFromCacheReq) (resp *pbCache.ReduceBlackUserFromCacheResp, err error) {
return nil, nil log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp = &pbCache.ReduceBlackUserFromCacheResp{CommonResp: &pbCache.CommonResp{}}
if err := db.DB.ReduceBlackUserFromCache(req.UserID, req.BlackUserID); err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error())
resp.CommonResp.ErrCode = constant.ErrDB.ErrCode
resp.CommonResp.ErrMsg = constant.ErrDB.ErrMsg
return resp, nil
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
return resp, nil
} }

56
internal/rpc/friend/firend.go
Unescape View File

@ -10,6 +10,7 @@ import (
"Open_IM/pkg/common/token_verify" "Open_IM/pkg/common/token_verify"
cp "Open_IM/pkg/common/utils" cp "Open_IM/pkg/common/utils"
"Open_IM/pkg/grpc-etcdv3/getcdv3" "Open_IM/pkg/grpc-etcdv3/getcdv3"
pbCache "Open_IM/pkg/proto/cache"
pbFriend "Open_IM/pkg/proto/friend" pbFriend "Open_IM/pkg/proto/friend"
sdkws "Open_IM/pkg/proto/sdk_ws" sdkws "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
@ -83,6 +84,18 @@ func (s *friendServer) AddBlacklist(ctx context.Context, req *pbFriend.AddBlackl
return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil
} }
log.NewInfo(req.CommID.OperationID, "AddBlacklist rpc ok ", req.CommID.FromUserID, req.CommID.ToUserID) log.NewInfo(req.CommID.OperationID, "AddBlacklist rpc ok ", req.CommID.FromUserID, req.CommID.ToUserID)
reqAddBlackUserToCache := &pbCache.AddBlackUserToCacheReq{UserID: req.CommID.FromUserID, BlackUserID: req.CommID.ToUserID, OperationID: req.CommID.OperationID}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
cacheClient := pbCache.NewCacheClient(etcdConn)
cacheResp, err := cacheClient.AddBlackUserToCache(context.Background(), reqAddBlackUserToCache)
if err != nil {
log.NewError(req.CommID.OperationID, "AddBlackUserToCache rpc call failed ", err.Error())
return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: 500, ErrMsg: "AddBlackUserToCache rpc call failed"}}, nil
}
if cacheResp.CommonResp.ErrCode != 0 {
log.NewError(req.CommID.OperationID, "AddBlackUserToCache rpc logic call failed ", cacheResp.String())
return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}, nil
}
chat.BlackAddedNotification(req) chat.BlackAddedNotification(req)
return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{}}, nil return &pbFriend.AddBlacklistResp{CommonResp: &pbFriend.CommonResp{}}, nil
} }
@ -244,6 +257,24 @@ func (s *friendServer) AddFriendResponse(ctx context.Context, req *pbFriend.AddF
} else { } else {
log.Error(req.CommID.OperationID, "HandleResult failed ", req.HandleResult) log.Error(req.CommID.OperationID, "HandleResult failed ", req.HandleResult)
} }
addFriendToCacheReq := &pbCache.AddFriendToCacheReq{OperationID: req.CommID.OperationID}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
client := pbCache.NewCacheClient(etcdConn)
addFriendToCacheReq.UserID = req.CommID.ToUserID
addFriendToCacheReq.UserID = req.CommID.FromUserID
respPb, err := client.AddFriendToCache(context.Background(), addFriendToCacheReq)
addFriendToCacheReq.UserID = req.CommID.FromUserID
addFriendToCacheReq.UserID = req.CommID.ToUserID
respPb, err = client.AddFriendToCache(context.Background(), addFriendToCacheReq)
if err != nil {
log.NewError(req.CommID.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error())
return &pbFriend.AddFriendResponseResp{CommonResp: &pbFriend.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: constant.ErrServer.ErrMsg}}, nil
}
if respPb.CommonResp.ErrCode != 0 {
log.NewError(req.CommID.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed")
return &pbFriend.AddFriendResponseResp{CommonResp: &pbFriend.CommonResp{ErrCode: respPb.CommonResp.ErrCode, ErrMsg: respPb.CommonResp.ErrMsg}}, nil
}
log.NewInfo(req.CommID.OperationID, "rpc AddFriendResponse ok") log.NewInfo(req.CommID.OperationID, "rpc AddFriendResponse ok")
return &pbFriend.AddFriendResponseResp{CommonResp: &pbFriend.CommonResp{}}, nil return &pbFriend.AddFriendResponseResp{CommonResp: &pbFriend.CommonResp{}}, nil
} }
@ -262,6 +293,19 @@ func (s *friendServer) DeleteFriend(ctx context.Context, req *pbFriend.DeleteFri
} }
log.NewInfo(req.CommID.OperationID, "DeleteFriend rpc ok") log.NewInfo(req.CommID.OperationID, "DeleteFriend rpc ok")
chat.FriendDeletedNotification(req) chat.FriendDeletedNotification(req)
reduceFriendFromCache := &pbCache.ReduceFriendFromCacheReq{OperationID: req.CommID.OperationID, UserID: req.CommID.FromUserID, FriendID: req.CommID.ToUserID}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
client := pbCache.NewCacheClient(etcdConn)
respPb, err := client.ReduceFriendFromCache(context.Background(), reduceFriendFromCache)
if err != nil {
log.NewError(req.CommID.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error())
return &pbFriend.DeleteFriendResp{CommonResp: &pbFriend.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: constant.ErrServer.ErrMsg}}, nil
}
if respPb.CommonResp.ErrCode != 0 {
log.NewError(req.CommID.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed")
return &pbFriend.DeleteFriendResp{CommonResp: &pbFriend.CommonResp{ErrCode: respPb.CommonResp.ErrCode, ErrMsg: respPb.CommonResp.ErrMsg}}, nil
}
return &pbFriend.DeleteFriendResp{CommonResp: &pbFriend.CommonResp{}}, nil return &pbFriend.DeleteFriendResp{CommonResp: &pbFriend.CommonResp{}}, nil
} }
@ -330,6 +374,18 @@ func (s *friendServer) RemoveBlacklist(ctx context.Context, req *pbFriend.Remove
} }
log.NewInfo(req.CommID.OperationID, "rpc RemoveBlacklist ok ") log.NewInfo(req.CommID.OperationID, "rpc RemoveBlacklist ok ")
reqReduceBlackUserFromCache := &pbCache.ReduceBlackUserFromCacheReq{UserID: req.CommID.FromUserID, BlackUserID: req.CommID.ToUserID, OperationID: req.CommID.OperationID}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
cacheClient := pbCache.NewCacheClient(etcdConn)
cacheResp, err := cacheClient.ReduceBlackUserFromCache(context.Background(), reqReduceBlackUserFromCache)
if err != nil {
log.NewError(req.CommID.OperationID, "ReduceBlackUserFromCache rpc call failed ", err.Error())
return &pbFriend.RemoveBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: 500, ErrMsg: "ReduceBlackUserFromCache rpc call failed"}}, nil
}
if cacheResp.CommonResp.ErrCode != 0 {
log.NewError(req.CommID.OperationID, "ReduceBlackUserFromCache rpc logic call failed ", cacheResp.String())
return &pbFriend.RemoveBlacklistResp{CommonResp: &pbFriend.CommonResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}, nil
}
chat.BlackDeletedNotification(req) chat.BlackDeletedNotification(req)
return &pbFriend.RemoveBlacklistResp{CommonResp: &pbFriend.CommonResp{}}, nil return &pbFriend.RemoveBlacklistResp{CommonResp: &pbFriend.CommonResp{}}, nil
} }

52
internal/rpc/msg/send_msg.go
Unescape View File

@ -6,9 +6,9 @@ import (
"Open_IM/pkg/common/db" "Open_IM/pkg/common/db"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/grpc-etcdv3/getcdv3" "Open_IM/pkg/grpc-etcdv3/getcdv3"
cacheRpc "Open_IM/pkg/proto/cache"
pbChat "Open_IM/pkg/proto/chat" pbChat "Open_IM/pkg/proto/chat"
pbConversation "Open_IM/pkg/proto/conversation" pbConversation "Open_IM/pkg/proto/conversation"
rpc "Open_IM/pkg/proto/friend"
pbGroup "Open_IM/pkg/proto/group" pbGroup "Open_IM/pkg/proto/group"
sdk_ws "Open_IM/pkg/proto/sdk_ws" sdk_ws "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
@ -48,36 +48,38 @@ func userRelationshipVerification(data *pbChat.SendMsgReq) (bool, int32, string)
return true, 0, "" return true, 0, ""
} }
log.NewDebug(data.OperationID, config.Config.MessageVerify.FriendVerify) log.NewDebug(data.OperationID, config.Config.MessageVerify.FriendVerify)
req := &rpc.IsInBlackListReq{CommID: &rpc.CommID{}} reqGetBlackIDListFromCache := &cacheRpc.GetBlackIDListFromCacheReq{UserID: data.MsgData.RecvID, OperationID: data.OperationID}
req.CommID.OperationID = data.OperationID etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
req.CommID.OpUserID = data.MsgData.RecvID cacheClient := cacheRpc.NewCacheClient(etcdConn)
req.CommID.FromUserID = data.MsgData.RecvID cacheResp, err := cacheClient.GetBlackIDListFromCache(context.Background(), reqGetBlackIDListFromCache)
req.CommID.ToUserID = data.MsgData.SendID
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImFriendName)
client := rpc.NewFriendClient(etcdConn)
reply, err := client.IsInBlackList(context.Background(), req)
if err != nil { if err != nil {
log.NewDebug(data.OperationID, "IsInBlackListReq rpc failed, ", req.String(), err.Error()) log.NewError(data.OperationID, "GetBlackIDListFromCache rpc call failed ", err.Error())
} else if reply.Response == true { } else {
log.NewDebug(data.OperationID, "IsInBlackListReq ", req.String()) if cacheResp.CommonResp.ErrCode != 0 {
return false, 600, "in black list" log.NewError(data.OperationID, "GetBlackIDListFromCache rpc logic call failed ", cacheResp.String())
} else {
if utils.IsContain(data.MsgData.SendID, cacheResp.UserIDList) {
return false, 600, "in black list"
}
}
} }
log.NewDebug(data.OperationID, config.Config.MessageVerify.FriendVerify) log.NewDebug(data.OperationID, config.Config.MessageVerify.FriendVerify)
if config.Config.MessageVerify.FriendVerify { if config.Config.MessageVerify.FriendVerify {
friendReq := &rpc.IsFriendReq{CommID: &rpc.CommID{}} reqGetFriendIDListFromCache := &cacheRpc.GetFriendIDListFromCacheReq{UserID: data.MsgData.RecvID, OperationID: data.OperationID}
friendReq.CommID.OperationID = data.OperationID etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
friendReq.CommID.OpUserID = data.MsgData.RecvID cacheClient := cacheRpc.NewCacheClient(etcdConn)
friendReq.CommID.FromUserID = data.MsgData.RecvID cacheResp, err := cacheClient.GetFriendIDListFromCache(context.Background(), reqGetFriendIDListFromCache)
friendReq.CommID.ToUserID = data.MsgData.SendID
friendReply, err := client.IsFriend(context.Background(), friendReq)
if err != nil { if err != nil {
log.NewDebug(data.OperationID, "IsFriendReq rpc failed, ", req.String(), err.Error()) log.NewError(data.OperationID, "GetFriendIDListFromCache rpc call failed ", err.Error())
return true, 0, "" } else {
} else if friendReply.Response == false { if cacheResp.CommonResp.ErrCode != 0 {
log.NewDebug(data.OperationID, "not friend ", req.String()) log.NewError(data.OperationID, "GetFriendIDListFromCache rpc logic call failed ", cacheResp.String())
return friendReply.Response, 601, "not friend" } else {
if !utils.IsContain(data.MsgData.SendID, cacheResp.UserIDList) {
return false, 601, "not friend"
}
}
} }
log.NewDebug(data.OperationID, config.Config.MessageVerify.FriendVerify, friendReply.Response)
return true, 0, "" return true, 0, ""
} else { } else {
return true, 0, "" return true, 0, ""

18
pkg/base_info/user_api_struct.go
Unescape View File

@ -32,3 +32,21 @@ type GetSelfUserInfoResp struct {
UserInfo *open_im_sdk.UserInfo `json:"-"` UserInfo *open_im_sdk.UserInfo `json:"-"`
Data map[string]interface{} `json:"data"` Data map[string]interface{} `json:"data"`
} }
type GetFriendIDListFromCacheReq struct {
OperationID string `json:"operationID" binding:"required"`
}
type GetFriendIDListFromCacheResp struct {
CommResp
UserIDList []string `json:"userIDList" binding:"required"`
}
type GetBlackIDListFromCacheReq struct {
OperationID string `json:"operationID" binding:"required"`
}
type GetBlackIDListFromCacheResp struct {
CommResp
UserIDList []string `json:"userIDList" binding:"required"`
}

13
pkg/common/db/mysql_model/im_mysql_model/user_black_list_model.go
Unescape View File

@ -56,3 +56,16 @@ func GetBlackListByUserID(ownerUserID string) ([]db.Black, error) {
} }
return blackListUsersInfo, nil return blackListUsersInfo, nil
} }
func GetBlackIDListByUserID(ownerUserID string) ([]string, error) {
dbConn, err := db.DB.MysqlDB.DefaultGormDB()
if err != nil {
return nil, err
}
var blackIDList []string
err = dbConn.Table("blacks").Where("owner_user_id=?", ownerUserID).Pluck("block_user_id", &blackIDList).Error
if err != nil {
return nil, err
}
return blackIDList, nil
}

10
pkg/common/db/mysql_model/im_mysql_model/user_model.go
Unescape View File

@ -60,6 +60,16 @@ func DeleteUser(userID string) (i int64) {
return i return i
} }
func GetAllUser() ([]db.User, error) {
dbConn, err := db.DB.MysqlDB.DefaultGormDB()
if err != nil {
return nil, err
}
var userList []db.User
err = dbConn.Table("users").Find(&userList).Error
return userList, err
}
func GetUserByUserID(userID string) (*db.User, error) { func GetUserByUserID(userID string) (*db.User, error) {
dbConn, err := db.DB.MysqlDB.DefaultGormDB() dbConn, err := db.DB.MysqlDB.DefaultGormDB()
if err != nil { if err != nil {

82
pkg/common/db/redisModel.go
Unescape View File

@ -9,15 +9,17 @@ import (
) )
const ( const (
AccountTempCode = "ACCOUNT_TEMP_CODE" accountTempCode = "ACCOUNT_TEMP_CODE"
resetPwdTempCode = "RESET_PWD_TEMP_CODE" resetPwdTempCode = "RESET_PWD_TEMP_CODE"
userIncrSeq = "REDIS_USER_INCR_SEQ:" // user incr seq userIncrSeq = "REDIS_USER_INCR_SEQ:" // user incr seq
appleDeviceToken = "DEVICE_TOKEN" appleDeviceToken = "DEVICE_TOKEN"
userMinSeq = "REDIS_USER_MIN_SEQ:" userMinSeq = "REDIS_USER_MIN_SEQ:"
uidPidToken = "UID_PID_TOKEN_STATUS:" uidPidToken = "UID_PID_TOKEN_STATUS:"
conversationReceiveMessageOpt = "CON_RECV_MSG_OPT:" conversationReceiveMessageOpt = "CON_RECV_MSG_OPT:"
GetuiToken = "GETUI" getuiToken = "GETUI"
UserInfoCache = "USER_INFO_CACHE:" userInfoCache = "USER_INFO_CACHE:"
friendRelationCache = "FRIEND_RELATION_CACHE:"
blackListCache = "BLACK_LIST_CACHE:"
) )
func (d *DataBases) Exec(cmd string, key interface{}, args ...interface{}) (interface{}, error) { func (d *DataBases) Exec(cmd string, key interface{}, args ...interface{}) (interface{}, error) {
@ -40,16 +42,16 @@ func (d *DataBases) Exec(cmd string, key interface{}, args ...interface{}) (inte
return con.Do(cmd, params...) return con.Do(cmd, params...)
} }
func (d *DataBases) JudgeAccountEXISTS(account string) (bool, error) { func (d *DataBases) JudgeAccountEXISTS(account string) (bool, error) {
key := AccountTempCode + account key := accountTempCode + account
return redis.Bool(d.Exec("EXISTS", key)) return redis.Bool(d.Exec("EXISTS", key))
} }
func (d *DataBases) SetAccountCode(account string, code, ttl int) (err error) { func (d *DataBases) SetAccountCode(account string, code, ttl int) (err error) {
key := AccountTempCode + account key := accountTempCode + account
_, err = d.Exec("SET", key, code, "ex", ttl) _, err = d.Exec("SET", key, code, "ex", ttl)
return err return err
} }
func (d *DataBases) GetAccountCode(account string) (string, error) { func (d *DataBases) GetAccountCode(account string) (string, error) {
key := AccountTempCode + account key := accountTempCode + account
return redis.String(d.Exec("GET", key)) return redis.String(d.Exec("GET", key))
} }
@ -150,27 +152,27 @@ func (d *DataBases) GetMultiConversationMsgOpt(userID string, conversationIDs []
} }
func (d *DataBases) SetGetuiToken(token string, expireTime int64) error { func (d *DataBases) SetGetuiToken(token string, expireTime int64) error {
_, err := d.Exec("SET", GetuiToken, token, "ex", expireTime) _, err := d.Exec("SET", getuiToken, token, "ex", expireTime)
return err return err
} }
func (d *DataBases) GetGetuiToken() (string, error) { func (d *DataBases) GetGetuiToken() (string, error) {
result, err := redis.String(d.Exec("GET", GetuiToken)) result, err := redis.String(d.Exec("GET", getuiToken))
return result, err return result, err
} }
func (d *DataBases) SetUserInfo(userInfo *pbCommon.UserInfo) error { func (d *DataBases) SearchContentType() {
b, _ := json.Marshal(&userInfo)
m := map[string]interface{}{} }
if err := json.Unmarshal(b, &m); err != nil {
return err func (d *DataBases) SetUserInfoToCache(userID string, m map[string]interface{}) error {
} _, err := d.Exec("hmset", userInfoCache+userID, redis.Args{}.Add().AddFlat(m)...)
_, err := d.Exec("hmset", UserInfoCache+userInfo.UserID, redis.Args{}.Add().AddFlat(m)...)
return err return err
} }
func (d *DataBases) GetUserInfo(userID string) (*pbCommon.UserInfo, error) { func (d *DataBases) GetUserInfoFromCache(userID string) (*pbCommon.UserInfo, error) {
result, err := redis.String(d.Exec("HGETALL", UserInfoCache+userID)) result, err := redis.String(d.Exec("hgetall", userInfoCache+userID))
log2.NewInfo("", result)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -178,3 +180,49 @@ func (d *DataBases) GetUserInfo(userID string) (*pbCommon.UserInfo, error) {
err = json.Unmarshal([]byte(result), userInfo) err = json.Unmarshal([]byte(result), userInfo)
return userInfo, err return userInfo, err
} }
func (d *DataBases) AddFriendToCache(userID string, friendIDList ...string) error {
var IDList []interface{}
for _, id := range friendIDList {
IDList = append(IDList, id)
}
_, err := d.Exec("SADD", friendRelationCache+userID, IDList...)
return err
}
func (d *DataBases) ReduceFriendToCache(userID string, friendIDList ...string) error {
var IDList []interface{}
for _, id := range friendIDList {
IDList = append(IDList, id)
}
_, err := d.Exec("SREM", friendRelationCache+userID, IDList...)
return err
}
func (d *DataBases) GetFriendIDListFromCache(userID string) ([]string, error) {
result, err := redis.Strings(d.Exec("SMEMBERS", friendRelationCache+userID))
return result, err
}
func (d *DataBases) AddBlackUserToCache(userID string, blackList ...string) error {
var IDList []interface{}
for _, id := range blackList {
IDList = append(IDList, id)
}
_, err := d.Exec("SADD", blackListCache+userID, IDList...)
return err
}
func (d *DataBases) ReduceBlackUserFromCache(userID string, blackList ...string) error {
var IDList []interface{}
for _, id := range blackList {
IDList = append(IDList, id)
}
_, err := d.Exec("SREM", blackListCache+userID, IDList...)
return err
}
func (d *DataBases) GetBlackListFromCache(userID string) ([]string, error) {
result, err := redis.Strings(d.Exec("SMEMBERS", blackListCache+userID))
return result, err
}

5
pkg/common/utils/utils.go
Unescape View File

@ -154,8 +154,3 @@ func UserDBCopyOpenIM(dst *open_im_sdk.UserInfo, src *db.User) {
func UserDBCopyOpenIMPublicUser(dst *open_im_sdk.PublicUserInfo, src *db.User) { func UserDBCopyOpenIMPublicUser(dst *open_im_sdk.PublicUserInfo, src *db.User) {
utils.CopyStructFields(dst, src) utils.CopyStructFields(dst, src)
} }
//
//func PublicUserDBCopyOpenIM(dst *open_im_sdk.PublicUserInfo, src *db.User){
//
//}

1030
pkg/proto/cache/cache.pb.go vendored
View File

File diff suppressed because it is too large Load Diff

79
pkg/proto/cache/cache.proto vendored
Unescape View File

@ -8,71 +8,100 @@ message CommonResp{
string errMsg = 2; string errMsg = 2;
} }
message GetUserInfoReq{ message GetUserInfoFromCacheReq{
repeated string userIDList = 1; repeated string userIDList = 1;
string operationID = 3; string operationID = 2;
} }
message GetUserInfoResp{ message GetUserInfoFromCacheResp{
CommonResp commonResp = 1; CommonResp commonResp = 1;
repeated server_api_params.UserInfo userInfoList = 2; repeated server_api_params.UserInfo userInfoList = 2;
} }
message UpdateUserInfoReq{ message UpdateUserInfoToCacheReq{
repeated server_api_params.UserInfo userInfoList = 1; repeated server_api_params.UserInfo userInfoList = 1;
string operationID = 2; string operationID = 2;
} }
message UpdateUserInfoResp{ message UpdateUserInfoToCacheResp{
CommonResp commonResp = 1; CommonResp commonResp = 1;
} }
message GetFriendInfoReq { message GetFriendIDListFromCacheReq {
string userID = 1; string userID = 1;
string operationID = 2; string operationID = 2;
} }
message GetFriendInfoResp { message GetFriendIDListFromCacheResp {
repeated server_api_params.FriendInfo friendInfoList = 1; repeated string userIDList = 1;
CommonResp commonResp = 2; CommonResp commonResp = 2;
} }
message UpdateBlackListReq { message AddFriendToCacheReq {
repeated server_api_params.BlackInfo blackList = 1; string userID = 1;
string operationID = 2; string friendID = 2;
string operationID = 3;
}
message AddFriendToCacheResp {
CommonResp commonResp = 1;
} }
message UpdateBlackListResp { message ReduceFriendFromCacheReq {
string userID = 1;
string friendID = 2;
string operationID = 3;
}
message ReduceFriendFromCacheResp {
CommonResp commonResp = 1; CommonResp commonResp = 1;
} }
message GetBlackListReq { message GetBlackIDListFromCacheReq {
string userID = 1; string userID = 1;
string operationID = 2; string operationID = 2;
} }
message GetBlackListResp { message GetBlackIDListFromCacheResp {
repeated server_api_params.BlackInfo blackList = 1; repeated string userIDList = 1;
CommonResp commonResp = 2;
}
message AddBlackUserToCacheReq {
string userID = 1;
string blackUserID = 2;
string operationID = 3;
} }
message AddBlackUserToCacheResp {
CommonResp commonResp = 1;
}
message UpdateFriendInfoReq{ message ReduceBlackUserFromCacheReq {
repeated server_api_params.FriendInfo friendInfoList = 1; string userID = 1;
string operationID = 2; string blackUserID = 2;
string operationID = 3;
} }
message UpdateFriendInfoResp{ message ReduceBlackUserFromCacheResp {
CommonResp commonResp = 1; CommonResp commonResp = 1;
} }
service cache{ service cache{
rpc GetUserInfo(GetUserInfoReq) returns(GetUserInfoResp); // userInfo
rpc UpdateUserInfo(UpdateUserInfoReq) returns(UpdateUserInfoResp); rpc GetUserInfoFromCache(GetUserInfoFromCacheReq) returns(GetUserInfoFromCacheResp);
rpc GetFriendInfo(GetFriendInfoReq) returns(GetFriendInfoResp); rpc UpdateUserInfoToCache(UpdateUserInfoToCacheReq) returns(UpdateUserInfoToCacheResp);
rpc UpdateFriendInfo(UpdateFriendInfoReq) returns(UpdateFriendInfoResp);
rpc UpdateBlackList(UpdateBlackListReq) returns(UpdateBlackListResp); // friendInfo
rpc GetBlackList(GetBlackListReq) returns(GetBlackListResp); rpc GetFriendIDListFromCache(GetFriendIDListFromCacheReq) returns(GetFriendIDListFromCacheResp);
rpc AddFriendToCache(AddFriendToCacheReq) returns(AddFriendToCacheResp);
rpc ReduceFriendFromCache(ReduceFriendFromCacheReq) returns(ReduceFriendFromCacheResp);
// blackList
rpc GetBlackIDListFromCache(GetBlackIDListFromCacheReq) returns(GetBlackIDListFromCacheResp);
rpc AddBlackUserToCache(AddBlackUserToCacheReq) returns(AddBlackUserToCacheResp);
rpc ReduceBlackUserFromCache(ReduceBlackUserFromCacheReq) returns(ReduceBlackUserFromCacheResp);
} }

19
pkg/utils/utils.go
Unescape View File

@ -1,6 +1,10 @@
package utils package utils
import ( import (
"bytes"
"encoding/json"
"github.com/gogo/protobuf/jsonpb"
"github.com/golang/protobuf/proto"
"github.com/jinzhu/copier" "github.com/jinzhu/copier"
"github.com/pkg/errors" "github.com/pkg/errors"
"math/rand" "math/rand"
@ -131,3 +135,18 @@ func RemoveRepeatedStringInList(slc []string) []string {
} }
return result return result
} }
func Pb2Map(pb proto.Message) (map[string]interface{}, error) {
_buffer := bytes.Buffer{}
jsonbMarshaller := &jsonpb.Marshaler{
OrigName: true,
EnumsAsInts: true,
EmitDefaults: true,
}
_ = jsonbMarshaller.Marshal(&_buffer, pb)
jsonCnt := _buffer.Bytes()
var out map[string]interface{}
err := json.Unmarshal(jsonCnt, &out)
return out, err
}

Write Preview
Loading…
Cancel
Save