cache

config/config.yaml

@@ -143,8 +143,8 @@ rpcregistername: #rpc注册服务名，默认即可
   openImOnlineMessageRelayName: OnlineMessageRelay
   openImGroupName: Group
   openImAuthName: Auth
-  OpenImStatisticsName: Statistics
+  openImStatisticsName: Statistics
-  OpenImMessageCMSName: MessageCMS
+  openImMessageCMSName: MessageCMS
   openImAdminCMSName: AdminCMS
   openImOfficeName: Office
   openImOrganizationName: Organization

internal/msg_gateway/gate/validate.go

@@ -114,140 +114,3 @@ func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, er
 	//} else
 
 }
-
-//func (ws *WServer) signalMessageAssemble(s *open_im_sdk.SignalReq, operationID string) (isPass bool, errCode int32, errMsg string, r *open_im_sdk.SignalResp, msgData *open_im_sdk.MsgData) {
-//	var msg open_im_sdk.MsgData
-//	var resp open_im_sdk.SignalResp
-//	media := open_im_media.NewMedia()
-//	msg.MsgFrom = constant.UserMsgType
-//	msg.ContentType = constant.SignalingNotification
-//	reqData, e := proto.Marshal(s)
-//	if e != nil {
-//		return false, 201, e.Error(), nil, nil
-//	}
-//	msg.Content = reqData
-//	msg.CreateTime = utils.GetCurrentTimestampByMill()
-//	options := make(map[string]bool, 6)
-//	utils.SetSwitchFromOptions(options, constant.IsHistory, false)
-//	utils.SetSwitchFromOptions(options, constant.IsPersistent, false)
-//	utils.SetSwitchFromOptions(options, constant.IsSenderSync, true)
-//	utils.SetSwitchFromOptions(options, constant.IsConversationUpdate, false)
-//	utils.SetSwitchFromOptions(options, constant.IsSenderConversationUpdate, false)
-//	utils.SetSwitchFromOptions(options, constant.IsUnreadCount, false)
-//	utils.SetSwitchFromOptions(options, constant.IsOfflinePush, true)
-//	msg.Options = options
-//	switch payload := s.Payload.(type) {
-//	case *open_im_sdk.SignalReq_Invite:
-//		token, liveURL, err2 := media.GetJoinToken(payload.Invite.Invitation.RoomID, payload.Invite.Invitation.InviterUserID, operationID, payload.Invite.Participant)
-//		if err2 != nil {
-//			return false, 202, err2.Error(), nil, nil
-//		}
-//		invite := open_im_sdk.SignalResp_Invite{&open_im_sdk.SignalInviteReply{
-//			Token:   token,
-//			RoomID:  payload.Invite.Invitation.RoomID,
-//			LiveURL: liveURL,
-//		}}
-//		resp.Payload = &invite
-//		msg.SenderPlatformID = payload.Invite.Invitation.PlatformID
-//		msg.SessionType = payload.Invite.Invitation.SessionType
-//		msg.OfflinePushInfo = payload.Invite.OfflinePushInfo
-//		msg.SendID = payload.Invite.Invitation.InviterUserID
-//		if len(payload.Invite.Invitation.InviteeUserIDList) > 0 {
-//			msg.RecvID = payload.Invite.Invitation.InviteeUserIDList[0]
-//		} else {
-//			return false, 203, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//		}
-//		msg.ClientMsgID = utils.GetMsgID(payload.Invite.Invitation.InviterUserID)
-//		return true, 0, "", &resp, &msg
-//	case *open_im_sdk.SignalReq_InviteInGroup:
-//		token, liveURL, err2 := media.GetJoinToken(payload.InviteInGroup.Invitation.RoomID, payload.InviteInGroup.Invitation.InviterUserID, operationID, payload.InviteInGroup.Participant)
-//		if err2 != nil {
-//			return false, 204, err2.Error(), nil, nil
-//		}
-//		inviteGroup := open_im_sdk.SignalResp_InviteInGroup{&open_im_sdk.SignalInviteInGroupReply{
-//			RoomID:  payload.InviteInGroup.Invitation.RoomID,
-//			Token:   token,
-//			LiveURL: liveURL,
-//		}}
-//		resp.Payload = &inviteGroup
-//		msg.SenderPlatformID = payload.InviteInGroup.Invitation.PlatformID
-//		msg.SessionType = payload.InviteInGroup.Invitation.SessionType
-//		msg.OfflinePushInfo = payload.InviteInGroup.OfflinePushInfo
-//		msg.SendID = payload.InviteInGroup.Invitation.InviterUserID
-//		if len(payload.InviteInGroup.Invitation.InviteeUserIDList) > 0 {
-//			msg.GroupID = payload.InviteInGroup.Invitation.GroupID
-//		} else {
-//			return false, 205, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//		}
-//		msg.ClientMsgID = utils.GetMsgID(payload.InviteInGroup.Invitation.InviterUserID)
-//
-//		return true, 0, "", &resp, &msg
-//	case *open_im_sdk.SignalReq_Cancel:
-//		cancel := open_im_sdk.SignalResp_Cancel{&open_im_sdk.SignalCancelReply{}}
-//		resp.Payload = &cancel
-//		msg.OfflinePushInfo = payload.Cancel.OfflinePushInfo
-//		msg.SendID = payload.Cancel.Invitation.InviterUserID
-//		msg.SenderPlatformID = payload.Cancel.Invitation.PlatformID
-//		msg.SessionType = payload.Cancel.Invitation.SessionType
-//		if len(payload.Cancel.Invitation.InviteeUserIDList) > 0 {
-//			switch payload.Cancel.Invitation.SessionType {
-//			case constant.SingleChatType:
-//				msg.RecvID = payload.Cancel.Invitation.InviteeUserIDList[0]
-//			case constant.GroupChatType:
-//				msg.GroupID = payload.Cancel.Invitation.GroupID
-//			}
-//		} else {
-//			return false, 206, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//		}
-//		msg.ClientMsgID = utils.GetMsgID(payload.Cancel.OpUserID)
-//		return true, 0, "", &resp, &msg
-//	case *open_im_sdk.SignalReq_Accept:
-//		token, liveURL, err2 := media.GetJoinToken(payload.Accept.Invitation.RoomID, payload.Accept.OpUserID, operationID, payload.Accept.Participant)
-//		if err2 != nil {
-//			return false, 207, err2.Error(), nil, nil
-//		}
-//		accept := open_im_sdk.SignalResp_Accept{&open_im_sdk.SignalAcceptReply{
-//			Token:   token,
-//			LiveURL: liveURL,
-//			RoomID:  payload.Accept.Invitation.RoomID,
-//		}}
-//		resp.Payload = &accept
-//		msg.OfflinePushInfo = payload.Accept.OfflinePushInfo
-//		msg.SendID = payload.Accept.OpUserID
-//		msg.SenderPlatformID = payload.Accept.Invitation.PlatformID
-//		msg.SessionType = payload.Accept.Invitation.SessionType
-//		if len(payload.Accept.Invitation.InviteeUserIDList) > 0 {
-//			switch payload.Accept.Invitation.SessionType {
-//			case constant.SingleChatType:
-//				msg.RecvID = payload.Accept.Invitation.InviterUserID
-//			case constant.GroupChatType:
-//				msg.GroupID = payload.Accept.Invitation.GroupID
-//			}
-//		} else {
-//			return false, 208, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//		}
-//		msg.ClientMsgID = utils.GetMsgID(payload.Accept.OpUserID)
-//		return true, 0, "", &resp, &msg
-//	case *open_im_sdk.SignalReq_HungUp:
-//	case *open_im_sdk.SignalReq_Reject:
-//		reject := open_im_sdk.SignalResp_Reject{&open_im_sdk.SignalRejectReply{}}
-//		resp.Payload = &reject
-//		msg.OfflinePushInfo = payload.Reject.OfflinePushInfo
-//		msg.SendID = payload.Reject.OpUserID
-//		msg.SenderPlatformID = payload.Reject.Invitation.PlatformID
-//		msg.SessionType = payload.Reject.Invitation.SessionType
-//		if len(payload.Reject.Invitation.InviteeUserIDList) > 0 {
-//			switch payload.Reject.Invitation.SessionType {
-//			case constant.SingleChatType:
-//				msg.RecvID = payload.Reject.Invitation.InviterUserID
-//			case constant.GroupChatType:
-//				msg.GroupID = payload.Reject.Invitation.GroupID
-//			}
-//		} else {
-//			return false, 209, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//		}
-//		msg.ClientMsgID = utils.GetMsgID(payload.Reject.OpUserID)
-//		return true, 0, "", &resp, &msg
-//	}
-//	return false, 210, errors.New("InviteeUserIDList is null").Error(), nil, nil
-//}

internal/rpc/group/group.go

@@ -11,6 +11,7 @@ import (
 	"Open_IM/pkg/common/token_verify"
 	cp "Open_IM/pkg/common/utils"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
+	pbCache "Open_IM/pkg/proto/cache"
 	pbGroup "Open_IM/pkg/proto/group"
 	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
 	pbUser "Open_IM/pkg/proto/user"
@@ -136,12 +137,12 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
 		}
 
 		okUserIDList = append(okUserIDList, user.UserID)
-		err = db.DB.AddGroupMember(groupId, user.UserID)
+		// mongoDB method
-		if err != nil {
+		//err = db.DB.AddGroupMember(groupId, user.UserID)
-			log.NewError(req.OperationID, "add mongo group member failed, db.DB.AddGroupMember failed ", err.Error())
+		//if err != nil {
-		}
+		//	log.NewError(req.OperationID, "add mongo group member failed, db.DB.AddGroupMember failed ", err.Error())
+		//}
 	}
-
 	resp := &pbGroup.CreateGroupResp{GroupInfo: &open_im_sdk.GroupInfo{}}
 	group, err := imdb.GetGroupInfoByGroupID(groupId)
 	if err != nil {
@@ -160,8 +161,24 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
 	}
 	resp.GroupInfo.OwnerUserID = req.OwnerUserID
 
-	log.NewInfo(req.OperationID, "rpc CreateGroup return ", resp.String())
+	addGroupMemberToCacheReq := &pbCache.AddGroupMemberToCacheReq{
+		UserIDList:  okUserIDList,
+		GroupID:     req.GroupInfo.GroupID,
+		OperationID: req.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.AddGroupMemberToCache(context.Background(), addGroupMemberToCacheReq)
+	if err != nil {
+		log.NewError(req.OperationID, "AddBlackUserToCache rpc call failed ", err.Error())
+		return &pbGroup.CreateGroupResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
+	if cacheResp.CommonResp.ErrCode != 0 {
+		log.NewError(req.OperationID, "AddBlackUserToCache rpc logic call failed ", cacheResp.String())
+		return &pbGroup.CreateGroupResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
 
+	log.NewInfo(req.OperationID, "rpc CreateGroup return ", resp.String())
 	chat.GroupCreatedNotification(req.OperationID, req.OpUserID, groupId, okUserIDList)
 	return resp, nil
 }
@@ -307,6 +324,24 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
 			log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "SetConversation success", respPb.String(), v)
 		}
 	}
+
+	addGroupMemberToCacheReq := &pbCache.AddGroupMemberToCacheReq{
+		UserIDList:  okUserIDList,
+		GroupID:     req.GroupID,
+		OperationID: req.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.AddGroupMemberToCache(context.Background(), addGroupMemberToCacheReq)
+	if err != nil {
+		log.NewError(req.OperationID, "AddBlackUserToCache rpc call failed ", err.Error())
+		return &pbGroup.InviteUserToGroupResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
+	if cacheResp.CommonResp.ErrCode != 0 {
+		log.NewError(req.OperationID, "AddBlackUserToCache rpc logic call failed ", cacheResp.String())
+		return &pbGroup.InviteUserToGroupResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
+
 	chat.MemberInvitedNotification(req.OperationID, req.GroupID, req.OpUserID, req.Reason, okUserIDList)
 	resp.ErrCode = 0
 	log.NewInfo(req.OperationID, "InviteUserToGroup rpc return ", resp.String())
@@ -450,6 +485,24 @@ func (s *groupServer) KickGroupMember(ctx context.Context, req *pbGroup.KickGrou
 			log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "SetConversation success", respPb.String(), v)
 		}
 	}
+
+	reduceGroupMemberFromCacheReq := &pbCache.ReduceGroupMemberFromCacheReq{
+		UserIDList:  okUserIDList,
+		GroupID:     req.GroupID,
+		OperationID: req.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.ReduceGroupMemberFromCache(context.Background(), reduceGroupMemberFromCacheReq)
+	if err != nil {
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc call failed ", err.Error())
+		return &pbGroup.KickGroupMemberResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
+	if cacheResp.CommonResp.ErrCode != 0 {
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc logic call failed ", cacheResp.String())
+		return &pbGroup.KickGroupMemberResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}, nil
+	}
+
 	chat.MemberKickedNotification(req, okUserIDList)
 	log.NewInfo(req.OperationID, "GetGroupMemberList rpc return ", resp.String())
 	return &resp, nil
@@ -691,22 +744,24 @@ func (s *groupServer) QuitGroup(ctx context.Context, req *pbGroup.QuitGroupReq)
 	} else {
 		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "SetConversation success", respPb.String())
 	}
-	//reduceGroupMemberFromCacheReq := &pbCache.ReduceGroupMemberFromCacheReq{
+
-	//	UserIDList:           []string{req.OpUserID},
+	reduceGroupMemberFromCacheReq := &pbCache.ReduceGroupMemberFromCacheReq{
-	//	GroupID:              req.GroupID,
+		UserIDList:  []string{req.OpUserID},
-	//	OperationID:          req.OperationID,
+		GroupID:     req.GroupID,
-	//}
+		OperationID: req.OperationID,
-	//etcdConn = getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
+	}
-	//cacheClient := pbCache.NewCacheClient(etcdConn)
+	etcdConnCache := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
-	//respCachePb, err := cacheClient.ReduceGroupMemberFromCache(context.Background(), reduceGroupMemberFromCacheReq)
+	cacheClient := pbCache.NewCacheClient(etcdConnCache)
-	//if err != nil {
+	cacheResp, err := cacheClient.ReduceGroupMemberFromCache(context.Background(), reduceGroupMemberFromCacheReq)
-	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", err.Error(), reduceGroupMemberFromCacheReq.String())
+	if err != nil {
-	//	return &pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: constant.ErrServer.ErrMsg}}, nil
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc call failed ", err.Error())
-	//}
+		return &pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil
-	//if respPb.CommonResp.ErrCode != 0 {
+	}
-	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), "AddFriendToCache failed", reduceGroupMemberFromCacheReq.String())
+	if cacheResp.CommonResp.ErrCode != 0 {
-	//	return &pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode:respCachePb.CommonResp.ErrCode, ErrMsg: respCachePb.CommonResp.ErrMsg}}, nil
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc logic call failed ", cacheResp.String())
-	//}
+		return &pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}}, nil
+	}
+
 	chat.MemberQuitNotification(req)
 	log.NewInfo(req.OperationID, "rpc QuitGroup return ", pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: 0, ErrMsg: ""}})
 	return &pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: 0, ErrMsg: ""}}, nil
@@ -1047,9 +1102,26 @@ func (s *groupServer) RemoveGroupMembersCMS(_ context.Context, req *pbGroup.Remo
 			log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "SetConversation success", respPb.String(), v)
 		}
 	}
+
+	reduceGroupMemberFromCacheReq := &pbCache.ReduceGroupMemberFromCacheReq{
+		UserIDList:  resp.Success,
+		GroupID:     req.GroupId,
+		OperationID: req.OperationID,
+	}
+	etcdConnCache := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
+	cacheClient := pbCache.NewCacheClient(etcdConnCache)
+	cacheResp, err := cacheClient.ReduceGroupMemberFromCache(context.Background(), reduceGroupMemberFromCacheReq)
+	if err != nil {
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc call failed ", err.Error())
+		return resp, http.WrapError(constant.ErrDB)
+	}
+	if cacheResp.CommonResp.ErrCode != 0 {
+		log.NewError(req.OperationID, "ReduceGroupMemberFromCache rpc logic call failed ", cacheResp.String())
+		return resp, http.WrapError(constant.ErrDB)
+	}
+
 	chat.MemberKickedNotification(reqKick, resp.Success)
-	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "success: ", resp.Success)
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp)
-	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "failed: ", resp.Failed)
 	return resp, nil
 }
 
@@ -1086,6 +1158,24 @@ func (s *groupServer) AddGroupMembersCMS(_ context.Context, req *pbGroup.AddGrou
 			resp.Success = append(resp.Success, userId)
 		}
 	}
+
+	addGroupMemberToCacheReq := &pbCache.AddGroupMemberToCacheReq{
+		UserIDList:  resp.Success,
+		GroupID:     req.GroupId,
+		OperationID: req.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.AddGroupMemberToCache(context.Background(), addGroupMemberToCacheReq)
+	if err != nil {
+		log.NewError(req.OperationId, "AddBlackUserToCache rpc call failed ", err.Error())
+		return resp, http.WrapError(constant.ErrDB)
+	}
+	if cacheResp.CommonResp.ErrCode != 0 {
+		log.NewError(req.OperationId, "AddBlackUserToCache rpc logic call failed ", cacheResp.String())
+		return resp, http.WrapError(constant.ErrDB)
+	}
+
 	chat.MemberInvitedNotification(req.OperationId, req.GroupId, req.OpUserId, "admin add you to group", resp.Success)
 	return resp, nil
 }

internal/rpc/msg/callback.go

@@ -31,6 +31,7 @@ func callbackBeforeSendSingleMsg(msg *pbChat.SendMsgReq) (canSend bool, err erro
 	if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable {
 		return true, nil
 	}
+
 	log.NewDebug(msg.OperationID, utils.GetSelfFuncName(), msg)
 	commonCallbackReq := copyCallbackCommonReqStruct(msg)
 	commonCallbackReq.CallbackCommand = constant.CallbackBeforeSendSingleMsgCommand

pkg/common/config/config.go

@@ -106,7 +106,7 @@ type config struct {
 		RpcGetTokenPort       []int `yaml:"rpcGetTokenPort"`
 	}
 	RpcRegisterName struct {
-		OpenImStatisticsName         string `yaml:"OpenImStatisticsName"`
+		OpenImStatisticsName         string `yaml:"openImStatisticsName"`
 		OpenImUserName               string `yaml:"openImUserName"`
 		OpenImFriendName             string `yaml:"openImFriendName"`
 		OpenImOfflineMessageName     string `yaml:"openImOfflineMessageName"`