diff --git a/cmd/rpc/open_im_cache/Makefile b/cmd/rpc/open_im_cache/Makefile new file mode 100644 index 000000000..1cb1e959e --- /dev/null +++ b/cmd/rpc/open_im_cache/Makefile @@ -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 + diff --git a/cmd/rpc/open_im_cache/main.go b/cmd/rpc/open_im_cache/main.go new file mode 100644 index 000000000..2ab2eefa8 --- /dev/null +++ b/cmd/rpc/open_im_cache/main.go @@ -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() + +} diff --git a/config/config.yaml b/config/config.yaml index 4bb653511..770fc31f6 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -124,6 +124,7 @@ rpcport: #rpc服务端口 默认即可 openImOfficePort: [ 10210 ] openImOrganizationPort: [ 10220 ] openImConversationPort: [ 10230 ] + openImCachePort: [10240] c2c: callbackBeforeSendMsg: switch: false diff --git a/internal/api/user/user.go b/internal/api/user/user.go index b7cb4c3df..af8448522 100644 --- a/internal/api/user/user.go +++ b/internal/api/user/user.go @@ -8,7 +8,6 @@ import ( "Open_IM/pkg/common/log" "Open_IM/pkg/common/token_verify" "Open_IM/pkg/grpc-etcdv3/getcdv3" - "Open_IM/pkg/proto/cache" pbRelay "Open_IM/pkg/proto/relay" open_im_sdk "Open_IM/pkg/proto/sdk_ws" rpc "Open_IM/pkg/proto/user" @@ -58,6 +57,14 @@ func GetUsersInfoFromCache(c *gin.Context) { c.JSON(http.StatusOK, resp) } +func GetUserFriendFromCache(c *gin.Context) { + +} + +func GetBlackListFromCache(c *gin.Context) { + +} + func GetUsersInfo(c *gin.Context) { params := api.GetUsersInfoReq{} if err := c.BindJSON(¶ms); err != nil { @@ -65,100 +72,34 @@ func GetUsersInfo(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"errCode": http.StatusBadRequest, "errMsg": err.Error()}) return } - getUserInfoReq := &rpc.GetUserInfoReq{} - getUserInfoReq.OperationID = params.OperationID + req := &rpc.GetUserInfoReq{} + utils.CopyStructFields(req, ¶ms) 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 { - 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"}) return } - log.NewInfo(params.OperationID, "GetUserInfo args ", getUserInfoReq.String()) - reqCacheGetUserInfo := &cache.GetUserInfoReq{} - utils.CopyStructFields(reqCacheGetUserInfo, ¶ms) - 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.NewInfo(params.OperationID, "GetUserInfo args ", req.String()) - log.NewDebug(getUserInfoReq.OperationID, utils.GetSelfFuncName(), "need cache user list", needCacheUserIDList) - getUserInfoReq.UserIDList = needCacheUserIDList - etcdConn = getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName) + 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(), getUserInfoReq) - if err != nil { - log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", err.Error(), getUserInfoReq.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) + RpcResp, err := client.GetUserInfo(context.Background(), req) if err != nil { - log.NewError(getUserInfoReq.OperationID, "GetUserInfo failed ", err.Error()) + log.NewError(req.OperationID, "GetUserInfo failed ", err.Error(), req.String()) c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed"}) return } - userInfoList = rpcResp.UserInfoList - for _, v := range userInfoList { + 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 := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList} 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) } @@ -181,34 +122,18 @@ func UpdateUserInfo(c *gin.Context) { return } 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) - rpcResp, err := client.UpdateUserInfo(context.Background(), req) + + etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName) + client := rpc.NewUserClient(etcdConn) + RpcResp, err := client.UpdateUserInfo(context.Background(), req) if err != nil { log.NewError(req.OperationID, "UpdateUserInfo failed ", err.Error(), req.String()) c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call rpc server failed"}) return } - if rpcResp.CommonResp.ErrCode != 0 { - log.NewError(req.OperationID, utils.GetSelfFuncName(), rpcResp.CommonResp.String()) - resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}} - 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) - } + resp := api.UpdateUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}} + log.NewInfo(req.OperationID, "UpdateUserInfo api return ", resp) + c.JSON(http.StatusOK, resp) } func GetSelfUserInfo(c *gin.Context) { @@ -240,31 +165,16 @@ func GetSelfUserInfo(c *gin.Context) { return } if len(RpcResp.UserInfoList) == 1 { - updateUserInfoReq := &cache.UpdateUserInfoReq{UserInfoList: []*open_im_sdk.UserInfo{RpcResp.UserInfoList[0]}} - 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) - 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) + 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 { resp := api.GetSelfUserInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}} log.NewInfo(req.OperationID, "GetUserInfo api return ", resp) c.JSON(http.StatusOK, resp) } + } func GetUsersOnlineStatus(c *gin.Context) { diff --git a/pkg/proto/cache/cache.proto b/pkg/proto/cache/cache.proto index cecfabd3c..c618c20f8 100644 --- a/pkg/proto/cache/cache.proto +++ b/pkg/proto/cache/cache.proto @@ -15,12 +15,12 @@ message GetUserInfoReq{ message GetUserInfoResp{ CommonResp commonResp = 1; - repeated server_api_params.UserInfo UserInfoList = 2; + repeated server_api_params.UserInfo userInfoList = 2; } message UpdateUserInfoReq{ - repeated server_api_params.UserInfo UserInfoList = 1; + repeated server_api_params.UserInfo userInfoList = 1; string operationID = 2; } @@ -28,17 +28,31 @@ message UpdateUserInfoResp{ CommonResp commonResp = 1; } -message UpdateAllUserToCacheReq{ - string operationID = 1; + +message GetFriendInfoReq { + } -message UpdateAllUserToCacheResp{ - CommonResp commonResp = 1; +message GetFriendInfoResp { + +} + + +message UpdateFriendInfoReq{ + repeated server_api_params.FriendInfo friendInfoList = 1; + string operationID = 2; } +message UpdateFriendInfoResp{ + CommonResp commonResp = 1; +} service cache{ rpc GetUserInfo(GetUserInfoReq) returns(GetUserInfoResp); rpc UpdateUserInfo(UpdateUserInfoReq) returns(UpdateUserInfoResp); - rpc UpdateAllUserToCache(UpdateAllUserToCacheReq) returns(UpdateAllUserToCacheResp); + rpc GetFriendInfo(GetFriendInfoReq) returns(GetFriendInfoResp); + rpc UpdateFriendInfo(UpdateFriendInfoReq) returns(UpdateFriendInfoResp); + rpc } + + diff --git a/script/check_all.sh b/script/check_all.sh index 60abfad69..ccd62684b 100644 --- a/script/check_all.sh +++ b/script/check_all.sh @@ -22,6 +22,7 @@ service_port_name=( openImOfficePort openImOrganizationPort openImConversationPort + openImCachePort ) switch=$(cat $config_path | grep demoswitch |awk -F '[:]' '{print $NF}') for i in ${service_port_name[*]}; do diff --git a/script/path_info.cfg b/script/path_info.cfg index 3071a0a5d..9c9b0c264 100644 --- a/script/path_info.cfg +++ b/script/path_info.cfg @@ -51,6 +51,7 @@ service_source_root=( ../cmd/rpc/open_im_office/ ../cmd/rpc/open_im_organization/ ../cmd/rpc/open_im_conversation/ + ../cmd/rpc/open_im_cache/ ${msg_gateway_source_root} ${msg_transfer_source_root} ${msg_source_root} @@ -74,6 +75,7 @@ service_names=( open_im_office open_im_organization open_im_conversation + open_im_cache ${msg_gateway_name} ${msg_transfer_name} ${msg_name} diff --git a/script/start_rpc_service.sh b/script/start_rpc_service.sh index 6e60bcdcd..351bd2797 100644 --- a/script/start_rpc_service.sh +++ b/script/start_rpc_service.sh @@ -21,6 +21,7 @@ service_filename=( open_im_office open_im_organization open_im_conversation + open_im_cache ) #service config port name @@ -40,6 +41,7 @@ service_port_name=( openImOfficePort openImOrganizationPort openImConversationPort + openImCachePort ) for ((i = 0; i < ${#service_filename[*]}; i++)); do