blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-10-08 01:55:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

cache rpc

Browse Source
This commit is contained in:
wangchuxiao 2022-04-26 10:33:49 +08:00
parent b211a66e99
commit 7e44f7f110
8 changed files with 100 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
cmd/rpc/open_im_cache/Makefile Normal file
View File

@ -0,0 +1,24 @@
.PHONY: all build run gotool install clean help
BINARY_NAME=open_im_cache
BIN_DIR=../../../bin/
all: gotool build
build:
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags="-w -s"
run:
@go run ./
gotool:
go fmt ./
go vet ./
install:
make build
mv ${BINARY_NAME} ${BIN_DIR}
clean:
@if [ -f ${BINARY_NAME} ] ; then rm ${BINARY_NAME} ; fi

16
cmd/rpc/open_im_cache/main.go Normal file
View File

@ -0,0 +1,16 @@
package main
import (
rpcCache "Open_IM/internal/rpc/cache"
"flag"
"fmt"
)
func main() {
rpcPort := flag.Int("port", 10600, "RpcToken default listen port 10800")
flag.Parse()
fmt.Println("start auth rpc server, port: ", *rpcPort)
rpcServer := rpcCache.NewOfficeServer(*rpcPort)
rpcServer.Run()
}

1
config/config.yaml
View File

@ -124,6 +124,7 @@ rpcport: #rpc服务端口 默认即可
openImOfficePort: [ 10210 ] openImOfficePort: [ 10210 ]
openImOrganizationPort: [ 10220 ] openImOrganizationPort: [ 10220 ]
openImConversationPort: [ 10230 ] openImConversationPort: [ 10230 ]
openImCachePort: [10240]
c2c: c2c:
callbackBeforeSendMsg: callbackBeforeSendMsg:
switch: false switch: false

156
internal/api/user/user.go
View File

@ -8,7 +8,6 @@ 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"
"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"
@ -58,6 +57,14 @@ func GetUsersInfoFromCache(c *gin.Context) {
c.JSON(http.StatusOK, resp) c.JSON(http.StatusOK, resp)
} }
func GetUserFriendFromCache(c *gin.Context) {
}
func GetBlackListFromCache(c *gin.Context) {
}
func GetUsersInfo(c *gin.Context) { func GetUsersInfo(c *gin.Context) {
params := api.GetUsersInfoReq{} params := api.GetUsersInfoReq{}
if err := c.BindJSON(&params); err != nil { if err := c.BindJSON(&params); err != nil {
@ -65,100 +72,34 @@ func GetUsersInfo(c *gin.Context) {
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
} }
getUserInfoReq := &rpc.GetUserInfoReq{} req := &rpc.GetUserInfoReq{}
getUserInfoReq.OperationID = params.OperationID utils.CopyStructFields(req, &params)
var ok bool var ok bool
ok, getUserInfoReq.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), getUserInfoReq.OperationID) ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok { if !ok {
log.NewError(getUserInfoReq.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 ", getUserInfoReq.String()) log.NewInfo(params.OperationID, "GetUserInfo args ", req.String())
reqCacheGetUserInfo := &cache.GetUserInfoReq{}
utils.CopyStructFields(reqCacheGetUserInfo, &params)
var userInfoList []*open_im_sdk.UserInfo
var publicUserInfoList []*open_im_sdk.PublicUserInfo
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
cacheClient := cache.NewCacheClient(etcdConn)
cacheResp, err := cacheClient.GetUserInfo(context.Background(), reqCacheGetUserInfo)
if err != nil {
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()})
return
}
if cacheResp.CommonResp.ErrCode != 0 {
log.NewError(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "GetUserInfo failed", cacheResp.CommonResp)
resp := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}
resp.Data = []map[string]interface{}{}
log.NewInfo(getUserInfoReq.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp)
return
}
log.NewInfo(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "cacheResp:", cacheResp.String())
userInfoList = cacheResp.UserInfoList
var needCacheUserIDList []string
for _, userID := range reqCacheGetUserInfo.UserIDList {
isGetUserInfoFromCache := false
for _, cacheUser := range userInfoList {
if cacheUser.UserID == userID {
isGetUserInfoFromCache = true
}
}
if !isGetUserInfoFromCache {
needCacheUserIDList = append(needCacheUserIDList, userID)
}
}
if len(needCacheUserIDList) == 0 {
log.NewInfo(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "get all userInfo from cache success")
for _, v := range 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: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList}
resp.Data = jsonData.JsonDataList(resp.UserInfoList)
log.NewInfo(getUserInfoReq.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp)
return
}
log.NewDebug(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "need cache user list", needCacheUserIDList) etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
getUserInfoReq.UserIDList = needCacheUserIDList
etcdConn = getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
client := rpc.NewUserClient(etcdConn) client := rpc.NewUserClient(etcdConn)
rpcResp, err := client.GetUserInfo(context.Background(), getUserInfoReq) RpcResp, err := client.GetUserInfo(context.Background(), req)
if err != nil { if err != nil {
log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", err.Error(), getUserInfoReq.String()) log.NewError(req.OperationID, "GetUserInfo failed ", err.Error(), req.String())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed" + err.Error()})
return
}
if rpcResp.CommonResp.ErrCode != 0 {
log.NewError(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "GetUserInfo failed", cacheResp.CommonResp)
resp := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}
resp.Data = []map[string]interface{}{}
log.NewInfo(getUserInfoReq.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp)
return
}
userInfoList = append(userInfoList, rpcResp.UserInfoList...)
cacheUpdateUserInfoReq := &cache.UpdateUserInfoReq{
UserInfoList: rpcResp.UserInfoList,
OperationID: getUserInfoReq.OperationID,
}
_, err = cacheClient.UpdateUserInfo(context.Background(), cacheUpdateUserInfoReq)
if err != nil {
log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", 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"})
return return
} }
userInfoList = rpcResp.UserInfoList var publicUserInfoList []*open_im_sdk.PublicUserInfo
for _, v := range userInfoList { for _, v := range RpcResp.UserInfoList {
publicUserInfoList = append(publicUserInfoList, publicUserInfoList = append(publicUserInfoList,
&open_im_sdk.PublicUserInfo{UserID: v.UserID, Nickname: v.Nickname, FaceURL: v.FaceURL, Gender: v.Gender, Ex: v.Ex}) &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 := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList}
resp.Data = jsonData.JsonDataList(resp.UserInfoList) resp.Data = jsonData.JsonDataList(resp.UserInfoList)
log.NewInfo(getUserInfoReq.OperationID, "GetUserInfo api return ", resp) log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp) c.JSON(http.StatusOK, resp)
} }
@ -181,34 +122,18 @@ func UpdateUserInfo(c *gin.Context) {
return return
} }
log.NewInfo(params.OperationID, "UpdateUserInfo args ", req.String()) log.NewInfo(params.OperationID, "UpdateUserInfo args ", req.String())
etcdConnUser := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
client := rpc.NewUserClient(etcdConnUser) etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
rpcResp, err := client.UpdateUserInfo(context.Background(), req) client := rpc.NewUserClient(etcdConn)
RpcResp, err := client.UpdateUserInfo(context.Background(), req)
if err != nil { if err != nil {
log.NewError(req.OperationID, "UpdateUserInfo failed ", err.Error(), req.String()) log.NewError(req.OperationID, "UpdateUserInfo failed ", err.Error(), req.String())
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"})
return return
} }
if rpcResp.CommonResp.ErrCode != 0 { resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}}
log.NewError(req.OperationID, utils.GetSelfFuncName(), rpcResp.CommonResp.String()) log.NewInfo(req.OperationID, "UpdateUserInfo api return ", resp)
resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}} c.JSON(http.StatusOK, resp)
c.JSON(http.StatusOK, resp)
return
} else {
updateUserInfoReq := &cache.UpdateUserInfoReq{UserInfoList: []*open_im_sdk.UserInfo{req.UserInfo}}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), updateUserInfoReq.String())
etcdConnCache := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
cacheClient := cache.NewCacheClient(etcdConnCache)
cacheResp, err := cacheClient.UpdateUserInfo(context.Background(), updateUserInfoReq)
if err != nil {
log.NewError(req.OperationID, "UpdateUserInfo cache failed ", err.Error(), req.String())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed: " + err.Error()})
return
}
resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}
log.NewInfo(req.OperationID, "UpdateUserInfo api return ", resp)
c.JSON(http.StatusOK, resp)
}
} }
func GetSelfUserInfo(c *gin.Context) { func GetSelfUserInfo(c *gin.Context) {
@ -240,31 +165,16 @@ func GetSelfUserInfo(c *gin.Context) {
return return
} }
if len(RpcResp.UserInfoList) == 1 { if len(RpcResp.UserInfoList) == 1 {
updateUserInfoReq := &cache.UpdateUserInfoReq{UserInfoList: []*open_im_sdk.UserInfo{RpcResp.UserInfoList[0]}} resp := api.GetSelfUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfo: RpcResp.UserInfoList[0]}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), updateUserInfoReq.String()) resp.Data = jsonData.JsonDataOne(resp.UserInfo)
etcdConnCache := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName) log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
cacheClient := cache.NewCacheClient(etcdConnCache) c.JSON(http.StatusOK, resp)
cacheClient.UpdateUserInfo(context.Background(), updateUserInfoReq)
//if err != nil {
// log.NewError(req.OperationID, "UpdateUserInfo cache failed ", err.Error(), req.String())
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed: " + err.Error()})
// return
//}
//if cacheResp.CommonResp.ErrCode != 0 {
// log.NewError(req.OperationID, utils.GetSelfFuncName(), cacheResp.CommonResp.String())
// resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: cacheResp.CommonResp.ErrCode, ErrMsg: cacheResp.CommonResp.ErrMsg}}
// c.JSON(http.StatusOK, resp)
// return
//}
//resp := api.GetSelfUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfo: RpcResp.UserInfoList[0]}
//resp.Data = jsonData.JsonDataOne(resp.UserInfo)
//log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
//c.JSON(http.StatusOK, resp)
} else { } else {
resp := api.GetSelfUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}} resp := api.GetSelfUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}}
log.NewInfo(req.OperationID, "GetUserInfo api return ", resp) log.NewInfo(req.OperationID, "GetUserInfo api return ", resp)
c.JSON(http.StatusOK, resp) c.JSON(http.StatusOK, resp)
} }
} }
func GetUsersOnlineStatus(c *gin.Context) { func GetUsersOnlineStatus(c *gin.Context) {

28
pkg/proto/cache/cache.proto vendored
View File

@ -15,12 +15,12 @@ message GetUserInfoReq{
message GetUserInfoResp{ message GetUserInfoResp{
CommonResp commonResp = 1; CommonResp commonResp = 1;
repeated server_api_params.UserInfo UserInfoList = 2; repeated server_api_params.UserInfo userInfoList = 2;
} }
message UpdateUserInfoReq{ message UpdateUserInfoReq{
repeated server_api_params.UserInfo UserInfoList = 1; repeated server_api_params.UserInfo userInfoList = 1;
string operationID = 2; string operationID = 2;
} }
@ -28,17 +28,31 @@ message UpdateUserInfoResp{
CommonResp commonResp = 1; CommonResp commonResp = 1;
} }
message UpdateAllUserToCacheReq{
string operationID = 1; message GetFriendInfoReq {
} }
message UpdateAllUserToCacheResp{ message GetFriendInfoResp {
}
message UpdateFriendInfoReq{
repeated server_api_params.FriendInfo friendInfoList = 1;
string operationID = 2;
}
message UpdateFriendInfoResp{
CommonResp commonResp = 1; CommonResp commonResp = 1;
} }
service cache{ service cache{
rpc GetUserInfo(GetUserInfoReq) returns(GetUserInfoResp); rpc GetUserInfo(GetUserInfoReq) returns(GetUserInfoResp);
rpc UpdateUserInfo(UpdateUserInfoReq) returns(UpdateUserInfoResp); rpc UpdateUserInfo(UpdateUserInfoReq) returns(UpdateUserInfoResp);
rpc UpdateAllUserToCache(UpdateAllUserToCacheReq) returns(UpdateAllUserToCacheResp); rpc GetFriendInfo(GetFriendInfoReq) returns(GetFriendInfoResp);
rpc UpdateFriendInfo(UpdateFriendInfoReq) returns(UpdateFriendInfoResp);
rpc
} }

1
script/check_all.sh
View File

@ -22,6 +22,7 @@ service_port_name=(
openImOfficePort openImOfficePort
openImOrganizationPort openImOrganizationPort
openImConversationPort openImConversationPort
openImCachePort
) )
switch=$(cat $config_path | grep demoswitch |awk -F '[:]' '{print $NF}') switch=$(cat $config_path | grep demoswitch |awk -F '[:]' '{print $NF}')
for i in ${service_port_name[*]}; do for i in ${service_port_name[*]}; do

2
script/path_info.cfg
View File

@ -51,6 +51,7 @@ service_source_root=(
../cmd/rpc/open_im_office/ ../cmd/rpc/open_im_office/
../cmd/rpc/open_im_organization/ ../cmd/rpc/open_im_organization/
../cmd/rpc/open_im_conversation/ ../cmd/rpc/open_im_conversation/
../cmd/rpc/open_im_cache/
${msg_gateway_source_root} ${msg_gateway_source_root}
${msg_transfer_source_root} ${msg_transfer_source_root}
${msg_source_root} ${msg_source_root}
@ -74,6 +75,7 @@ service_names=(
open_im_office open_im_office
open_im_organization open_im_organization
open_im_conversation open_im_conversation
open_im_cache
${msg_gateway_name} ${msg_gateway_name}
${msg_transfer_name} ${msg_transfer_name}
${msg_name} ${msg_name}

2
script/start_rpc_service.sh
View File

@ -21,6 +21,7 @@ service_filename=(
open_im_office open_im_office
open_im_organization open_im_organization
open_im_conversation open_im_conversation
open_im_cache
) )
#service config port name #service config port name
@ -40,6 +41,7 @@ service_port_name=(
openImOfficePort openImOfficePort
openImOrganizationPort openImOrganizationPort
openImConversationPort openImConversationPort
openImCachePort
) )
for ((i = 0; i < ${#service_filename[*]}; i++)); do for ((i = 0; i < ${#service_filename[*]}; i++)); do