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

deploy_k8s/dockerfiles/api.Dockerfile

@@ -0,0 +1,15 @@
+FROM ubuntu
+
+# 设置固定的项目路径
+ENV WORKDIR /bin
+ENV CONFIG_NAME $WORKDIR/config/config.yaml
+
+# 将可执行文件复制到目标目录
+ADD ../cmd/bin/open_im_api $WORKDIR/main
+
+# 创建用于挂载的几个目录，添加可执行权限
+RUN mkdir $WORKDIR/logs $WORKDIR/config $WORKDIR/db && \
+  chmod +x $WORKDIR/main
+
+WORKDIR $WORKDIR
+CMD ./main

deploy_k8s/script/build_push_all_images.sh

@@ -0,0 +1,35 @@
+#/bin/sh
+#service_source_root=(
+#  #api service file
+#  ../cmd/open_im_api/
+#  ../cmd/open_im_cms_api/
+#  #rpc service file
+#  ../cmd/rpc/open_im_user/
+#  ../cmd/rpc/open_im_friend/
+#  ../cmd/rpc/open_im_group/
+#  ../cmd/rpc/open_im_auth/
+#  ../cmd/rpc/open_im_admin_cms/
+#  ../cmd/rpc/open_im_message_cms/
+#  ../cmd/rpc/open_im_statistics/
+#  ../cmd/rpc/open_im_office/
+#  ../cmd/rpc/open_im_organization/
+#  ../cmd/rpc/open_im_conversation/
+#  ../cmd/rpc/open_im_cache/
+#  ../cmd/open_im_msg_gateway/
+#  ../cmd/open_im_msg_transfer/
+#  ../cmd/rpc/open_im_msg/
+#  ../cmd/open_im_push/
+#  ../cmd/Open-IM-SDK-Core/
+#  ../cmd/open_im_demo/
+#)
+#
+dockerfile_list=$(ls ../dockerfiles/)
+echo ${dockerfile_list}
+for dockerfile in $dockerfile_list
+do
+	echo "start to build images" $dockerfile
+	docker build -t $image . -f ${dockerfile}
+	echo "build ${dockerfile} ok"
+done
+echo ${#dockerfile_list[*]}
+

internal/api/office/tag.go

@@ -262,8 +262,8 @@ func GetTagSendLogs(c *gin.Context) {
 	client := pbOffice.NewOfficeServiceClient(etcdConn)
 	respPb, err := client.GetTagSendLogs(context.Background(), &reqPb)
 	if err != nil {
-		log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetUserTags failed", err.Error())
-		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "CreateTag rpc server failed" + err.Error()})
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetTagSendLogs failed", err.Error())
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetTagSendLogs rpc server failed" + err.Error()})
 		return
 	}
 	if err := utils.CopyStructFields(&resp.CommResp, respPb.CommonResp); err != nil {

internal/msg_transfer/logic/init.go

@@ -24,6 +24,8 @@ var (
 	singleMsgSuccessCount uint64
 	groupMsgCount         uint64
 	singleMsgFailedCount  uint64
+
+	singleMsgSuccessCountMutex sync.Mutex
 )
 
 func Init() {

internal/msg_transfer/logic/offline_history_msg_handler.go

@@ -53,7 +53,9 @@ func (mc *OfflineHistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey s
 				log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
 				return
 			}
+			singleMsgSuccessCountMutex.Lock()
 			singleMsgSuccessCount++
+			singleMsgSuccessCountMutex.Unlock()
 			log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
 		}
 		if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {

internal/msg_transfer/logic/online_history_msg_handler.go

@@ -86,7 +86,9 @@ func (och *OnlineHistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey s
 				log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
 				return
 			}
+			singleMsgSuccessCountMutex.Lock()
 			singleMsgSuccessCount++
+			singleMsgSuccessCountMutex.Unlock()
 			log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
 		}
 		if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {

internal/rpc/office/office.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unsafe"
 )
 
 type officeServer struct {
@@ -56,7 +57,13 @@ func (s *officeServer) Run() {
 	log.NewInfo("0", "listen network success, ", address, listener)
 	defer listener.Close()
 	//grpc server
-	srv := grpc.NewServer()
+	recvSize := 1024 * 1024 * 30
+	sendSize := 1024 * 1024 * 30
+	var options = []grpc.ServerOption{
+		grpc.MaxRecvMsgSize(recvSize),
+		grpc.MaxSendMsgSize(sendSize),
+	}
+	srv := grpc.NewServer(options...)
 	defer srv.GracefulStop()
 	//Service registers with etcd
 	pbOffice.RegisterOfficeServiceServer(srv, s)
@@ -207,6 +214,12 @@ func (s *officeServer) SendMsg2Tag(_ context.Context, req *pbOffice.SendMsg2TagR
 			userIDList = append(userIDList[:i], userIDList[i+1:]...)
 		}
 	}
+	if unsafe.Sizeof(userIDList) > 1024*1024 {
+		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "size", unsafe.Sizeof(userIDList))
+		resp.CommonResp.ErrMsg = constant.ErrSendLimit.ErrMsg
+		resp.CommonResp.ErrCode = constant.ErrSendLimit.ErrCode
+		return
+	}
 	log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "total userIDList result: ", userIDList)
 	user, err := imdb.GetUserByUserID(req.SendID)
 	if err != nil {
@@ -548,9 +561,18 @@ func (s *officeServer) GetWorkMomentByID(_ context.Context, req *pbOffice.GetWor
 	if !canSee {
 		log.NewError(req.OperationID, utils.GetSelfFuncName(), "workMoments not access to user", canSee, workMoment, req.OpUserID)
 	}
+
 	if err := utils.CopyStructFields(resp.WorkMoment, workMoment); err != nil {
 		log.NewError(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
 	}
+	user, err := imdb.GetUserByUserID(workMoment.UserID)
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetUserByUserID failed", err.Error())
+	}
+	if user != nil {
+		resp.WorkMoment.FaceURL = user.FaceURL
+		resp.WorkMoment.UserName = user.Nickname
+	}
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
 	return resp, nil
 }
@@ -573,6 +595,16 @@ func (s *officeServer) GetUserWorkMoments(_ context.Context, req *pbOffice.GetUs
 	if err := utils.CopyStructFields(&resp.WorkMoments, workMoments); err != nil {
 		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields failed", err.Error())
 	}
+	for _, v := range resp.WorkMoments {
+		user, err := imdb.GetUserByUserID(v.UserID)
+		if err != nil {
+			log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error())
+		}
+		if user != nil {
+			v.UserName = user.Nickname
+			v.FaceURL = user.FaceURL
+		}
+	}
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
 	return resp, nil
 }
@@ -590,6 +622,16 @@ func (s *officeServer) GetUserFriendWorkMoments(_ context.Context, req *pbOffice
 	if err := utils.CopyStructFields(&resp.WorkMoments, workMoments); err != nil {
 		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields failed", err.Error())
 	}
+	for _, v := range resp.WorkMoments {
+		user, err := imdb.GetUserByUserID(v.UserID)
+		if err != nil {
+			log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error())
+		}
+		if user != nil {
+			v.UserName = user.Nickname
+			v.FaceURL = user.FaceURL
+		}
+	}
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
 	return resp, nil
 }

internal/rpc/user/user.go

@@ -7,7 +7,6 @@ import (
 	"Open_IM/pkg/common/db"
 	imdb "Open_IM/pkg/common/db/mysql_model/im_mysql_model"
 	errors "Open_IM/pkg/common/http"
-
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/common/token_verify"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
@@ -410,7 +409,21 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbUser.UpdateUserI
 	if req.UserInfo.FaceURL != "" {
 		go s.SyncJoinedGroupMemberFaceURL(req.UserInfo.UserID, req.UserInfo.FaceURL, req.OperationID, req.OpUserID)
 	}
-
+	//updateUserInfoToCacheReq := &cache.UpdateUserInfoToCacheReq{
+	//	OperationID:  req.OperationID,
+	//	UserInfoList: []*sdkws.UserInfo{req.UserInfo},
+	//}
+	//cacheEtcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
+	//cacheClient := cache.NewCacheClient(cacheEtcdConn)
+	//resp, err := cacheClient.UpdateUserInfoToCache(context.Background(), updateUserInfoToCacheReq)
+	//if err != nil {
+	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), updateUserInfoToCacheReq.String())
+	//	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: err.Error()}}, nil
+	//}
+	//if resp.CommonResp.ErrCode != 0 {
+	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), resp.String())
+	//	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: resp.CommonResp.ErrMsg}}, nil
+	//}
 	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{}}, nil
 }
 

pkg/common/constant/error.go

@@ -49,11 +49,12 @@ var (
 	ErrTokenUnknown     = ErrInfo{705, TokenUnknownMsg.Error()}
 	ErrTokenKicked      = ErrInfo{706, TokenUserKickedMsg.Error()}
 
-	ErrAccess   = ErrInfo{ErrCode: 801, ErrMsg: AccessMsg.Error()}
-	ErrDB       = ErrInfo{ErrCode: 802, ErrMsg: DBMsg.Error()}
-	ErrArgs     = ErrInfo{ErrCode: 803, ErrMsg: ArgsMsg.Error()}
-	ErrStatus   = ErrInfo{ErrCode: 804, ErrMsg: StatusMsg.Error()}
-	ErrCallback = ErrInfo{ErrCode: 809, ErrMsg: CallBackMsg.Error()}
+	ErrAccess    = ErrInfo{ErrCode: 801, ErrMsg: AccessMsg.Error()}
+	ErrDB        = ErrInfo{ErrCode: 802, ErrMsg: DBMsg.Error()}
+	ErrArgs      = ErrInfo{ErrCode: 803, ErrMsg: ArgsMsg.Error()}
+	ErrStatus    = ErrInfo{ErrCode: 804, ErrMsg: StatusMsg.Error()}
+	ErrCallback  = ErrInfo{ErrCode: 809, ErrMsg: CallBackMsg.Error()}
+	ErrSendLimit = ErrInfo{ErrCode: 810, ErrMsg: "send msg limit, to many users"}
 )
 
 var (