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

internal/common/convert/convert.go

@@ -46,27 +46,27 @@ func (db *DBFriend) DB2PB(ctx context.Context, friends []*relation.FriendModel)
 		return nil, err
 	}
 	for _, v := range friends {
-		pbFriend := &sdk.FriendInfo{FriendUser: &sdk.UserInfo{}}
-		utils.CopyStructFields(pbFriend, users[v.OwnerUserID])
-		utils.CopyStructFields(pbFriend.FriendUser, users[v.FriendUserID])
-		pbFriend.CreateTime = v.CreateTime.Unix()
-		pbFriend.FriendUser.CreateTime = v.CreateTime.Unix()
+		pbfriend := &sdk.FriendInfo{FriendUser: &sdk.UserInfo{}}
+		utils.CopyStructFields(pbfriend, users[v.OwnerUserID])
+		utils.CopyStructFields(pbfriend.FriendUser, users[v.FriendUserID])
+		pbfriend.CreateTime = v.CreateTime.Unix()
+		pbfriend.FriendUser.CreateTime = v.CreateTime.Unix()
 	}
 	return
 }
 
 func (db *DBFriend) Convert(ctx context.Context) (*sdk.FriendInfo, error) {
-	pbFriend := &sdk.FriendInfo{FriendUser: &sdk.UserInfo{}}
-	utils.CopyStructFields(pbFriend, db)
+	pbfriend := &sdk.FriendInfo{FriendUser: &sdk.UserInfo{}}
+	utils.CopyStructFields(pbfriend, db)
 	user, err := db.userCheck.GetUsersInfo(ctx, db.FriendUserID)
 	if err != nil {
 		return nil, err
 	}
-	utils.CopyStructFields(pbFriend.FriendUser, user)
-	pbFriend.CreateTime = db.CreateTime.Unix()
+	utils.CopyStructFields(pbfriend.FriendUser, user)
+	pbfriend.CreateTime = db.CreateTime.Unix()
 
-	pbFriend.FriendUser.CreateTime = db.CreateTime.Unix()
-	return pbFriend, nil
+	pbfriend.FriendUser.CreateTime = db.CreateTime.Unix()
+	return pbfriend, nil
 }
 
 func (pb *PBFriend) Convert() (*relation.FriendModel, error) {

internal/rpc/auth/auth.go

@@ -2,68 +2,45 @@ package auth
 
 import (
 	"Open_IM/internal/common/check"
-	"Open_IM/internal/common/rpcserver"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/cache"
 	"Open_IM/pkg/common/db/controller"
+	"Open_IM/pkg/common/db/relation"
+	relationTb "Open_IM/pkg/common/db/table/relation"
 	"Open_IM/pkg/common/log"
-	promePkg "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
+	discoveryRegistry "Open_IM/pkg/discoveryregistry"
 	pbAuth "Open_IM/pkg/proto/auth"
 	pbRelay "Open_IM/pkg/proto/relay"
 	"Open_IM/pkg/utils"
 	"context"
-	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/OpenIMSDK/openKeeper"
 	"google.golang.org/grpc"
 )
 
-func NewRpcAuthServer(port int) *rpcAuth {
-	r, err := rpcserver.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImAuthName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
+func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
+	mysql, err := relation.NewGormDB()
 	if err != nil {
-		panic(err)
+		return err
 	}
-	var redis cache.RedisClient
-	redis.InitRedis()
-	return &rpcAuth{
-		RpcServer:     r,
-		AuthInterface: controller.NewAuthController(redis.GetClient(), config.Config.TokenPolicy.AccessSecret, config.Config.TokenPolicy.AccessExpire),
+	if err := mysql.AutoMigrate(&relationTb.FriendModel{}, &relationTb.FriendRequestModel{}, &relationTb.BlackModel{}); err != nil {
+		return err
 	}
+	redis, err := cache.NewRedis()
+	if err != nil {
+		return err
+	}
+	pbAuth.RegisterAuthServer(server, &authServer{
+		userCheck:      check.NewUserCheck(client),
+		RegisterCenter: client,
+		AuthInterface:  controller.NewAuthController(redis.GetClient(), config.Config.TokenPolicy.AccessSecret, config.Config.TokenPolicy.AccessExpire),
+	})
+	return nil
 }
 
-func (s *rpcAuth) Run() {
-	operationID := utils.OperationIDGenerator()
-	log.NewInfo(operationID, "rpc auth start...")
-	listener, address, err := rpcserver.GetTcpListen(config.Config.ListenIP, s.Port)
-	if err != nil {
-		panic(err)
-	}
-	log.NewInfo(operationID, "listen network success ", listener, address)
-	var grpcOpts []grpc.ServerOption
-	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
-		promePkg.NewUserRegisterCounter()
-		promePkg.NewUserLoginCounter()
-		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
-			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
-			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
-		}...)
-	}
-	srv := grpc.NewServer(grpcOpts...)
-	defer srv.GracefulStop()
-	pbAuth.RegisterAuthServer(srv, s)
-	err = srv.Serve(listener)
-	if err != nil {
-		panic(err)
-	}
-	log.NewInfo(operationID, "rpc auth ok")
-}
-
-func (s *rpcAuth) UserToken(ctx context.Context, req *pbAuth.UserTokenReq) (*pbAuth.UserTokenResp, error) {
+func (s *authServer) UserToken(ctx context.Context, req *pbAuth.UserTokenReq) (*pbAuth.UserTokenResp, error) {
 	resp := pbAuth.UserTokenResp{}
 	if _, err := s.userCheck.GetUsersInfo(ctx, req.UserID); err != nil {
 		return nil, err
@@ -77,7 +54,7 @@ func (s *rpcAuth) UserToken(ctx context.Context, req *pbAuth.UserTokenReq) (*pbA
 	return &resp, nil
 }
 
-func (s *rpcAuth) parseToken(ctx context.Context, tokensString string) (claims *tokenverify.Claims, err error) {
+func (s *authServer) parseToken(ctx context.Context, tokensString string) (claims *tokenverify.Claims, err error) {
 	claims, err = tokenverify.GetClaimFromToken(tokensString)
 	if err != nil {
 		return nil, utils.Wrap(err, "")
@@ -102,7 +79,7 @@ func (s *rpcAuth) parseToken(ctx context.Context, tokensString string) (claims *
 	return nil, constant.ErrTokenNotExist.Wrap()
 }
 
-func (s *rpcAuth) ParseToken(ctx context.Context, req *pbAuth.ParseTokenReq) (resp *pbAuth.ParseTokenResp, err error) {
+func (s *authServer) ParseToken(ctx context.Context, req *pbAuth.ParseTokenReq) (resp *pbAuth.ParseTokenResp, err error) {
 	resp = &pbAuth.ParseTokenResp{}
 	claims, err := s.parseToken(ctx, req.Token)
 	if err != nil {
@@ -114,7 +91,7 @@ func (s *rpcAuth) ParseToken(ctx context.Context, req *pbAuth.ParseTokenReq) (re
 	return resp, nil
 }
 
-func (s *rpcAuth) ForceLogout(ctx context.Context, req *pbAuth.ForceLogoutReq) (*pbAuth.ForceLogoutResp, error) {
+func (s *authServer) ForceLogout(ctx context.Context, req *pbAuth.ForceLogoutReq) (*pbAuth.ForceLogoutResp, error) {
 	resp := pbAuth.ForceLogoutResp{}
 	if err := tokenverify.CheckAdmin(ctx); err != nil {
 		return nil, err
@@ -125,7 +102,7 @@ func (s *rpcAuth) ForceLogout(ctx context.Context, req *pbAuth.ForceLogoutReq) (
 	return &resp, nil
 }
 
-func (s *rpcAuth) forceKickOff(ctx context.Context, userID string, platformID int32, operationID string) error {
+func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID int32, operationID string) error {
 	grpcCons, err := s.RegisterCenter.GetConns(config.Config.RpcRegisterName.OpenImRelayName)
 	if err != nil {
 		return err
@@ -140,8 +117,8 @@ func (s *rpcAuth) forceKickOff(ctx context.Context, userID string, platformID in
 	return constant.ErrInternalServer.Wrap()
 }
 
-type rpcAuth struct {
-	*rpcserver.RpcServer
+type authServer struct {
 	controller.AuthInterface
-	userCheck *check.UserCheck
+	userCheck      *check.UserCheck
+	RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
 }

internal/rpc/friend/friend.go

@@ -4,100 +4,49 @@ import (
 	"Open_IM/internal/common/check"
 	"Open_IM/internal/common/convert"
 	"Open_IM/internal/common/notification"
-	"Open_IM/internal/common/rpcserver"
-	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/controller"
 	"Open_IM/pkg/common/db/relation"
 	relationTb "Open_IM/pkg/common/db/table/relation"
-	"Open_IM/pkg/common/log"
-	"Open_IM/pkg/common/middleware"
-	promePkg "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
-	pbFriend "Open_IM/pkg/proto/friend"
+	discoveryRegistry "Open_IM/pkg/discoveryregistry"
+	pbfriend "Open_IM/pkg/proto/friend"
 	"Open_IM/pkg/utils"
 	"context"
-	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/OpenIMSDK/openKeeper"
 	"google.golang.org/grpc"
 )
 
 type friendServer struct {
-	*rpcserver.RpcServer
 	controller.FriendInterface
 	controller.BlackInterface
-	notification *notification.Check
-	userCheck    *check.UserCheck
+	notification   *notification.Check
+	userCheck      *check.UserCheck
+	RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
 }
 
-func NewFriendServer(port int) *friendServer {
-	r, err := rpcserver.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImFriendName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
+func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
+	mysql, err := relation.NewGormDB()
 	if err != nil {
-		panic(err)
+		return err
 	}
-	//mysql init
-	var mysql relation.Mysql
-	var model relation.FriendGorm
-	err = mysql.InitConn().AutoMigrateModel(&relationTb.FriendModel{})
-	if err != nil {
-		panic("db init err:" + err.Error())
-	}
-	err = mysql.InitConn().AutoMigrateModel(&relationTb.FriendRequestModel{})
-	if err != nil {
-		panic("db init err:" + err.Error())
-	}
-	err = mysql.InitConn().AutoMigrateModel(&relationTb.BlackModel{})
-	if err != nil {
-		panic("db init err:" + err.Error())
-	}
-	if mysql.GormConn() != nil {
-		model.DB = mysql.GormConn()
-	} else {
-		panic("db init err:" + "conn is nil")
-	}
-	return &friendServer{
-		RpcServer:       r,
-		FriendInterface: controller.NewFriendController(model.DB),
-		BlackInterface:  controller.NewBlackController(model.DB),
-	}
-}
-
-func (s *friendServer) Run() {
-	operationID := utils.OperationIDGenerator()
-	log.NewInfo(operationID, "friendServer run...")
-	listener, address, err := rpcserver.GetTcpListen(config.Config.ListenIP, s.Port)
-	if err != nil {
-		panic(err)
-	}
-
-	log.NewInfo(operationID, "listen ok ", address)
-	defer listener.Close()
-	//grpc server
-	var grpcOpts []grpc.ServerOption
-	grpcOpts = append(grpcOpts, grpc.UnaryInterceptor(middleware.RpcServerInterceptor))
-	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
-		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
-			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
-			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
-		}...)
-	}
-	srv := grpc.NewServer(grpcOpts...)
-	defer srv.GracefulStop()
-	pbFriend.RegisterFriendServer(srv, s)
-	err = srv.Serve(listener)
-	if err != nil {
-		log.NewError(operationID, "Serve failed ", err.Error(), listener)
-		return
+	if err := mysql.AutoMigrate(&relationTb.FriendModel{}, &relationTb.FriendRequestModel{}, &relationTb.BlackModel{}); err != nil {
+		return err
 	}
+	pbfriend.RegisterFriendServer(server, &friendServer{
+		FriendInterface: controller.NewFriendController(mysql),
+		BlackInterface:  controller.NewBlackController(mysql),
+		notification:    notification.NewCheck(client),
+		userCheck:       check.NewUserCheck(client),
+		RegisterCenter:  client,
+	})
+	return nil
 }
 
 // ok
-func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbFriend.ApplyToAddFriendReq) (resp *pbFriend.ApplyToAddFriendResp, err error) {
-	resp = &pbFriend.ApplyToAddFriendResp{}
+func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.ApplyToAddFriendReq) (resp *pbfriend.ApplyToAddFriendResp, err error) {
+	resp = &pbfriend.ApplyToAddFriendResp{}
 	if err := tokenverify.CheckAccessV3(ctx, req.FromUserID); err != nil {
 		return nil, err
 	}
@@ -125,8 +74,8 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbFriend.Apply
 }
 
 // ok
-func (s *friendServer) ImportFriends(ctx context.Context, req *pbFriend.ImportFriendReq) (resp *pbFriend.ImportFriendResp, err error) {
-	resp = &pbFriend.ImportFriendResp{}
+func (s *friendServer) ImportFriends(ctx context.Context, req *pbfriend.ImportFriendReq) (resp *pbfriend.ImportFriendResp, err error) {
+	resp = &pbfriend.ImportFriendResp{}
 	if err := tokenverify.CheckAdmin(ctx); err != nil {
 		return nil, err
 	}
@@ -148,8 +97,8 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *pbFriend.ImportFr
 }
 
 // ok
-func (s *friendServer) RespondFriendApply(ctx context.Context, req *pbFriend.RespondFriendApplyReq) (resp *pbFriend.RespondFriendApplyResp, err error) {
-	resp = &pbFriend.RespondFriendApplyResp{}
+func (s *friendServer) RespondFriendApply(ctx context.Context, req *pbfriend.RespondFriendApplyReq) (resp *pbfriend.RespondFriendApplyResp, err error) {
+	resp = &pbfriend.RespondFriendApplyResp{}
 	if err := s.userCheck.Access(ctx, req.ToUserID); err != nil {
 		return nil, err
 	}
@@ -174,8 +123,8 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *pbFriend.Res
 }
 
 // ok
-func (s *friendServer) DeleteFriend(ctx context.Context, req *pbFriend.DeleteFriendReq) (resp *pbFriend.DeleteFriendResp, err error) {
-	resp = &pbFriend.DeleteFriendResp{}
+func (s *friendServer) DeleteFriend(ctx context.Context, req *pbfriend.DeleteFriendReq) (resp *pbfriend.DeleteFriendResp, err error) {
+	resp = &pbfriend.DeleteFriendResp{}
 	if err := s.userCheck.Access(ctx, req.OwnerUserID); err != nil {
 		return nil, err
 	}
@@ -191,8 +140,8 @@ func (s *friendServer) DeleteFriend(ctx context.Context, req *pbFriend.DeleteFri
 }
 
 // ok
-func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbFriend.SetFriendRemarkReq) (resp *pbFriend.SetFriendRemarkResp, err error) {
-	resp = &pbFriend.SetFriendRemarkResp{}
+func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbfriend.SetFriendRemarkReq) (resp *pbfriend.SetFriendRemarkResp, err error) {
+	resp = &pbfriend.SetFriendRemarkResp{}
 	if err := s.userCheck.Access(ctx, req.OwnerUserID); err != nil {
 		return nil, err
 	}
@@ -208,8 +157,8 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbFriend.SetFri
 }
 
 // ok
-func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbFriend.GetDesignatedFriendsReq) (resp *pbFriend.GetDesignatedFriendsResp, err error) {
-	resp = &pbFriend.GetDesignatedFriendsResp{}
+func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbfriend.GetDesignatedFriendsReq) (resp *pbfriend.GetDesignatedFriendsResp, err error) {
+	resp = &pbfriend.GetDesignatedFriendsResp{}
 	if err := s.userCheck.Access(ctx, req.UserID); err != nil {
 		return nil, err
 	}
@@ -226,8 +175,8 @@ func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbFriend.G
 }
 
 // ok 获取接收到的好友申请（即别人主动申请的）
-func (s *friendServer) GetPaginationFriendsApplyTo(ctx context.Context, req *pbFriend.GetPaginationFriendsApplyToReq) (resp *pbFriend.GetPaginationFriendsApplyToResp, err error) {
-	resp = &pbFriend.GetPaginationFriendsApplyToResp{}
+func (s *friendServer) GetPaginationFriendsApplyTo(ctx context.Context, req *pbfriend.GetPaginationFriendsApplyToReq) (resp *pbfriend.GetPaginationFriendsApplyToResp, err error) {
+	resp = &pbfriend.GetPaginationFriendsApplyToResp{}
 	if err := s.userCheck.Access(ctx, req.UserID); err != nil {
 		return nil, err
 	}
@@ -244,8 +193,8 @@ func (s *friendServer) GetPaginationFriendsApplyTo(ctx context.Context, req *pbF
 }
 
 // ok 获取主动发出去的好友申请列表
-func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *pbFriend.GetPaginationFriendsApplyFromReq) (resp *pbFriend.GetPaginationFriendsApplyFromResp, err error) {
-	resp = &pbFriend.GetPaginationFriendsApplyFromResp{}
+func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *pbfriend.GetPaginationFriendsApplyFromReq) (resp *pbfriend.GetPaginationFriendsApplyFromResp, err error) {
+	resp = &pbfriend.GetPaginationFriendsApplyFromResp{}
 	if err := s.userCheck.Access(ctx, req.UserID); err != nil {
 		return nil, err
 	}
@@ -262,8 +211,8 @@ func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *p
 }
 
 // ok
-func (s *friendServer) IsFriend(ctx context.Context, req *pbFriend.IsFriendReq) (resp *pbFriend.IsFriendResp, err error) {
-	resp = &pbFriend.IsFriendResp{}
+func (s *friendServer) IsFriend(ctx context.Context, req *pbfriend.IsFriendReq) (resp *pbfriend.IsFriendResp, err error) {
+	resp = &pbfriend.IsFriendResp{}
 	resp.InUser1Friends, resp.InUser2Friends, err = s.FriendInterface.CheckIn(ctx, req.UserID1, req.UserID2)
 	if err != nil {
 		return nil, err
@@ -272,8 +221,8 @@ func (s *friendServer) IsFriend(ctx context.Context, req *pbFriend.IsFriendReq)
 }
 
 // ok
-func (s *friendServer) GetPaginationFriends(ctx context.Context, req *pbFriend.GetPaginationFriendsReq) (resp *pbFriend.GetPaginationFriendsResp, err error) {
-	resp = &pbFriend.GetPaginationFriendsResp{}
+func (s *friendServer) GetPaginationFriends(ctx context.Context, req *pbfriend.GetPaginationFriendsReq) (resp *pbfriend.GetPaginationFriendsResp, err error) {
+	resp = &pbfriend.GetPaginationFriendsResp{}
 	if utils.Duplicate(req.FriendUserIDs) {
 		return nil, constant.ErrArgs.Wrap("friend userID repeated")
 	}

internal/rpc/group/callback.go

@@ -12,6 +12,7 @@ import (
 	"Open_IM/pkg/utils"
 	"context"
 	"google.golang.org/protobuf/types/known/wrapperspb"
+	"time"
 )
 
 func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (err error) {
@@ -21,72 +22,44 @@ func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (
 	defer func() {
 		tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", req)
 	}()
-	operationID := tracelog.GetOperationID(ctx)
-	commonCallbackReq := &callbackstruct.CallbackBeforeCreateGroupReq{
+	cbReq := &callbackstruct.CallbackBeforeCreateGroupReq{
 		CallbackCommand: constant.CallbackBeforeCreateGroupCommand,
-		OperationID:     operationID,
+		OperationID:     tracelog.GetOperationID(ctx),
 		GroupInfo:       *req.GroupInfo,
 	}
-	commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+	cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
 		UserID:    req.OwnerUserID,
 		RoleLevel: constant.GroupOwner,
 	})
 	for _, userID := range req.AdminUserIDs {
-		commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+		cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
 			UserID:    userID,
 			RoleLevel: constant.GroupAdmin,
 		})
 	}
 	for _, userID := range req.AdminUserIDs {
-		commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
+		cbReq.InitMemberList = append(cbReq.InitMemberList, &apistruct.GroupAddMemberInfo{
 			UserID:    userID,
 			RoleLevel: constant.GroupOrdinaryUsers,
 		})
 	}
-	resp := &callbackstruct.CallbackBeforeCreateGroupResp{
-		CommonCallbackResp: &callbackstruct.CommonCallbackResp{OperationID: operationID},
-	}
-	err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeCreateGroupCommand, commonCallbackReq, resp, config.Config.Callback.CallbackBeforeCreateGroup)
+	resp := &callbackstruct.CallbackBeforeCreateGroupResp{}
+	err = http.CallBackPostReturnV2(config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeCreateGroup)
 	if err != nil {
 		return err
 	}
-
-	if resp.GroupID != nil {
-		req.GroupInfo.GroupID = *resp.GroupID
-	}
-	if resp.GroupName != nil {
-		req.GroupInfo.GroupName = *resp.GroupName
-	}
-	if resp.Notification != nil {
-		req.GroupInfo.Notification = *resp.Notification
-	}
-	if resp.Introduction != nil {
-		req.GroupInfo.Introduction = *resp.Introduction
-	}
-	if resp.FaceURL != nil {
-		req.GroupInfo.FaceURL = *resp.FaceURL
-	}
-	if resp.OwnerUserID != nil {
-		req.GroupInfo.OwnerUserID = *resp.OwnerUserID
-	}
-	if resp.Ex != nil {
-		req.GroupInfo.Ex = *resp.Ex
-	}
-	if resp.Status != nil {
-		req.GroupInfo.Status = *resp.Status
-	}
-	if resp.CreatorUserID != nil {
-		req.GroupInfo.CreatorUserID = *resp.CreatorUserID
-	}
-	if resp.GroupType != nil {
-		req.GroupInfo.GroupType = *resp.GroupType
-	}
-	if resp.NeedVerification != nil {
-		req.GroupInfo.NeedVerification = *resp.NeedVerification
-	}
-	if resp.LookMemberInfo != nil {
-		req.GroupInfo.LookMemberInfo = *resp.LookMemberInfo
-	}
+	utils.NotNilReplace(&req.GroupInfo.GroupID, resp.GroupID)
+	utils.NotNilReplace(&req.GroupInfo.GroupName, resp.GroupName)
+	utils.NotNilReplace(&req.GroupInfo.Notification, resp.Notification)
+	utils.NotNilReplace(&req.GroupInfo.Introduction, resp.Introduction)
+	utils.NotNilReplace(&req.GroupInfo.FaceURL, resp.FaceURL)
+	utils.NotNilReplace(&req.GroupInfo.OwnerUserID, resp.OwnerUserID)
+	utils.NotNilReplace(&req.GroupInfo.Ex, resp.Ex)
+	utils.NotNilReplace(&req.GroupInfo.Status, resp.Status)
+	utils.NotNilReplace(&req.GroupInfo.CreatorUserID, resp.CreatorUserID)
+	utils.NotNilReplace(&req.GroupInfo.GroupType, resp.GroupType)
+	utils.NotNilReplace(&req.GroupInfo.NeedVerification, resp.NeedVerification)
+	utils.NotNilReplace(&req.GroupInfo.LookMemberInfo, resp.LookMemberInfo)
 	return nil
 }
 
@@ -97,39 +70,26 @@ func CallbackBeforeMemberJoinGroup(ctx context.Context, groupMember *relation.Gr
 	defer func() {
 		tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "groupMember", *groupMember, "groupEx", groupEx)
 	}()
-	operationID := tracelog.GetOperationID(ctx)
-	callbackResp := callbackstruct.CommonCallbackResp{OperationID: operationID}
-	callbackReq := callbackstruct.CallbackBeforeMemberJoinGroupReq{
+	callbackReq := &callbackstruct.CallbackBeforeMemberJoinGroupReq{
 		CallbackCommand: constant.CallbackBeforeMemberJoinGroupCommand,
-		OperationID:     operationID,
+		OperationID:     tracelog.GetOperationID(ctx),
 		GroupID:         groupMember.GroupID,
 		UserID:          groupMember.UserID,
 		Ex:              groupMember.Ex,
 		GroupEx:         groupEx,
 	}
-	resp := &callbackstruct.CallbackBeforeMemberJoinGroupResp{
-		CommonCallbackResp: &callbackResp,
-	}
-	err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeMemberJoinGroupCommand, callbackReq,
-		resp, config.Config.Callback.CallbackBeforeMemberJoinGroup)
+	resp := &callbackstruct.CallbackBeforeMemberJoinGroupResp{}
+	err = http.CallBackPostReturnV2(config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeMemberJoinGroup)
 	if err != nil {
 		return err
 	}
 	if resp.MuteEndTime != nil {
-		groupMember.MuteEndTime = utils.UnixSecondToTime(*resp.MuteEndTime)
-	}
-	if resp.FaceURL != nil {
-		groupMember.FaceURL = *resp.FaceURL
-	}
-	if resp.Ex != nil {
-		groupMember.Ex = *resp.Ex
-	}
-	if resp.NickName != nil {
-		groupMember.Nickname = *resp.NickName
-	}
-	if resp.RoleLevel != nil {
-		groupMember.RoleLevel = *resp.RoleLevel
+		groupMember.MuteEndTime = time.UnixMilli(*resp.MuteEndTime)
 	}
+	utils.NotNilReplace(&groupMember.FaceURL, resp.FaceURL)
+	utils.NotNilReplace(&groupMember.Ex, resp.Ex)
+	utils.NotNilReplace(&groupMember.Nickname, resp.Nickname)
+	utils.NotNilReplace(&groupMember.RoleLevel, resp.RoleLevel)
 	return nil
 }
 
@@ -140,44 +100,40 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMe
 	defer func() {
 		tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", *req)
 	}()
-	operationID := tracelog.GetOperationID(ctx)
-	callbackResp := callbackstruct.CommonCallbackResp{OperationID: operationID}
 	callbackReq := callbackstruct.CallbackBeforeSetGroupMemberInfoReq{
 		CallbackCommand: constant.CallbackBeforeSetGroupMemberInfoCommand,
-		OperationID:     operationID,
+		OperationID:     tracelog.GetOperationID(ctx),
 		GroupID:         req.GroupID,
 		UserID:          req.UserID,
 	}
 	if req.Nickname != nil {
-		callbackReq.Nickname = req.Nickname.Value
+		callbackReq.Nickname = &req.Nickname.Value
 	}
 	if req.FaceURL != nil {
-		callbackReq.FaceURL = req.FaceURL.Value
+		callbackReq.FaceURL = &req.FaceURL.Value
 	}
 	if req.RoleLevel != nil {
-		callbackReq.RoleLevel = req.RoleLevel.Value
+		callbackReq.RoleLevel = &req.RoleLevel.Value
 	}
 	if req.Ex != nil {
-		callbackReq.Ex = req.Ex.Value
+		callbackReq.Ex = &req.Ex.Value
 	}
-	resp := &callbackstruct.CallbackBeforeSetGroupMemberInfoResp{
-		CommonCallbackResp: &callbackResp,
-	}
-	err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetGroupMemberInfoCommand, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo)
+	resp := &callbackstruct.CallbackBeforeSetGroupMemberInfoResp{}
+	err = http.CallBackPostReturnV2(config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo)
 	if err != nil {
 		return err
 	}
 	if resp.FaceURL != nil {
-		req.FaceURL = &wrapperspb.StringValue{Value: *resp.FaceURL}
+		req.FaceURL = wrapperspb.String(*resp.FaceURL)
 	}
 	if resp.Nickname != nil {
-		req.Nickname = &wrapperspb.StringValue{Value: *resp.Nickname}
+		req.Nickname = wrapperspb.String(*resp.Nickname)
 	}
 	if resp.RoleLevel != nil {
-		req.RoleLevel = &wrapperspb.Int32Value{Value: *resp.RoleLevel}
+		req.RoleLevel = wrapperspb.Int32(*resp.RoleLevel)
 	}
 	if resp.Ex != nil {
-		req.Ex = &wrapperspb.StringValue{Value: *resp.Ex}
+		req.Ex = wrapperspb.String(*resp.Ex)
 	}
-	return err
+	return nil
 }

internal/rpc/group/group.go

@@ -28,11 +28,11 @@ import (
 )
 
 func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
-	mysql, err := relation.NewGormDB()
+	db, err := relation.NewGormDB()
 	if err != nil {
 		return err
 	}
-	if err := mysql.AutoMigrate(&relationTb.GroupModel{}, &relationTb.GroupMemberModel{}, &relationTb.GroupRequestModel{}); err != nil {
+	if err := db.AutoMigrate(&relationTb.GroupModel{}, &relationTb.GroupMemberModel{}, &relationTb.GroupRequestModel{}); err != nil {
 		return err
 	}
 	redis, err := cache.NewRedis()
@@ -44,7 +44,7 @@ func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
 		return err
 	}
 	pbGroup.RegisterGroupServer(server, &groupServer{
-		GroupInterface:      controller.NewGroupInterface(mysql, redis.GetClient(), mongo.GetClient()),
+		GroupInterface:      controller.NewGroupInterface(controller.NewGroupDatabase(db, redis.GetClient(), mongo.GetClient())),
 		UserCheck:           check.NewUserCheck(client),
 		ConversationChecker: check.NewConversationChecker(client),
 	})

internal/rpc/msg/extend_msg_callback.go

@@ -1,7 +1,7 @@
 package msg
 
 import (
-	cbApi "Open_IM/pkg/call_back_struct"
+	cb "Open_IM/pkg/callbackstruct"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/http"
@@ -11,10 +11,10 @@ import (
 	http2 "net/http"
 )
 
-func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensionsReq) *cbApi.CallbackBeforeSetMessageReactionExtResp {
-	callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
+func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensionsReq) *cb.CallbackBeforeSetMessageReactionExtResp {
+	callbackResp := cb.CommonCallbackResp{OperationID: setReq.OperationID}
 	log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
-	req := cbApi.CallbackBeforeSetMessageReactionExtReq{
+	req := cb.CallbackBeforeSetMessageReactionExtReq{
 		OperationID:           setReq.OperationID,
 		CallbackCommand:       constant.CallbackBeforeSetMessageReactionExtensionCommand,
 		SourceID:              setReq.SourceID,
@@ -26,7 +26,7 @@ func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensio
 		IsExternalExtensions:  setReq.IsExternalExtensions,
 		MsgFirstModifyTime:    setReq.MsgFirstModifyTime,
 	}
-	resp := &cbApi.CallbackBeforeSetMessageReactionExtResp{CommonCallbackResp: &callbackResp}
+	resp := &cb.CallbackBeforeSetMessageReactionExtResp{CommonCallbackResp: &callbackResp}
 	defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp)
 	if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetMessageReactionExtensionCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
 		callbackResp.ErrCode = http2.StatusInternalServerError
@@ -36,8 +36,8 @@ func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensio
 
 }
 
-func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReactionExtensionsReq) *cbApi.CallbackDeleteMessageReactionExtResp {
-	callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
+func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReactionExtensionsReq) *cb.CallbackDeleteMessageReactionExtResp {
+	callbackResp := cb.CommonCallbackResp{OperationID: setReq.OperationID}
 	log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
 	req := cbApi.CallbackDeleteMessageReactionExtReq{
 		OperationID:           setReq.OperationID,
@@ -58,8 +58,8 @@ func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReacti
 	}
 	return resp
 }
-func callbackGetMessageListReactionExtensions(getReq *msg.GetMessageListReactionExtensionsReq) *cbApi.CallbackGetMessageListReactionExtResp {
-	callbackResp := cbApi.CommonCallbackResp{OperationID: getReq.OperationID}
+func callbackGetMessageListReactionExtensions(getReq *msg.GetMessageListReactionExtensionsReq) *cb.CallbackGetMessageListReactionExtResp {
+	callbackResp := cb.CommonCallbackResp{OperationID: getReq.OperationID}
 	log.NewDebug(getReq.OperationID, utils.GetSelfFuncName(), getReq.String())
 	req := cbApi.CallbackGetMessageListReactionExtReq{
 		OperationID:     getReq.OperationID,
@@ -78,8 +78,8 @@ func callbackGetMessageListReactionExtensions(getReq *msg.GetMessageListReaction
 	}
 	return resp
 }
-func callbackAddMessageReactionExtensions(setReq *msg.AddMessageReactionExtensionsReq) *cbApi.CallbackAddMessageReactionExtResp {
-	callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
+func callbackAddMessageReactionExtensions(setReq *msg.AddMessageReactionExtensionsReq) *cb.CallbackAddMessageReactionExtResp {
+	callbackResp := cb.CommonCallbackResp{OperationID: setReq.OperationID}
 	log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
 	req := cbApi.CallbackAddMessageReactionExtReq{
 		OperationID:           setReq.OperationID,
@@ -93,7 +93,7 @@ func callbackAddMessageReactionExtensions(setReq *msg.AddMessageReactionExtensio
 		IsExternalExtensions:  setReq.IsExternalExtensions,
 		MsgFirstModifyTime:    setReq.MsgFirstModifyTime,
 	}
-	resp := &cbApi.CallbackAddMessageReactionExtResp{CommonCallbackResp: &callbackResp}
+	resp := &cb.CallbackAddMessageReactionExtResp{CommonCallbackResp: &callbackResp}
 	defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp, *resp.CommonCallbackResp, resp.IsReact, resp.MsgFirstModifyTime)
 	if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackAddMessageListReactionExtensionsCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
 		callbackResp.ErrCode = http2.StatusInternalServerError

internal/rpc/msg/server.go

@@ -2,33 +2,23 @@ package msg
 
 import (
 	"Open_IM/internal/common/check"
-	"Open_IM/internal/common/notification"
-	"Open_IM/internal/common/rpcserver"
-	"Open_IM/pkg/common/config"
-	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/controller"
+	"Open_IM/pkg/common/db/relation"
+	tablerelation "Open_IM/pkg/common/db/table/relation"
+	discoveryRegistry "Open_IM/pkg/discoveryregistry"
+	"github.com/OpenIMSDK/openKeeper"
 
-	"Open_IM/pkg/common/kafka"
-	"Open_IM/pkg/common/log"
 	promePkg "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/proto/msg"
-	"Open_IM/pkg/utils"
-	"github.com/OpenIMSDK/getcdv3"
-	"net"
-	"strconv"
-	"strings"
-
-	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
-
 	"google.golang.org/grpc"
 )
 
 type msgServer struct {
-	*rpcserver.RpcServer
-	MsgInterface controller.MsgInterface
-	Group        *check.GroupChecker
-	User         *check.UserCheck
-	Conversation *check.ConversationChecker
+	RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
+	MsgInterface   controller.MsgInterface
+	Group          *check.GroupChecker
+	User           *check.UserCheck
+	Conversation   *check.ConversationChecker
 }
 
 type deleteMsg struct {
@@ -38,39 +28,34 @@ type deleteMsg struct {
 	OperationID string
 }
 
-func NewRpcChatServer(port int) *msgServer {
-	log.NewPrivateLog(constant.LogFileName)
-	rc := msgServer{
-		rpcPort:         port,
-		rpcRegisterName: config.Config.RpcRegisterName.OpenImMsgName,
-		etcdSchema:      config.Config.Etcd.EtcdSchema,
-		etcdAddr:        config.Config.Etcd.EtcdAddr,
-		dMessageLocker:  NewLockerMessage(),
+func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
+	mysql, err := relation.NewGormDB()
+	if err != nil {
+		return err
 	}
-	rc.messageWriter = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
-	//rc.offlineProducer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschatOffline.Addr, config.Config.Kafka.Ws2mschatOffline.Topic)
-	rc.delMsgCh = make(chan deleteMsg, 1000)
-	return &rc
+	if err := mysql.AutoMigrate(&tablerelation.UserModel{}); err != nil {
+		return err
+	}
+	s := &msgServer{
+		Conversation: check.NewConversationChecker(client),
+		User:         check.NewUserCheck(client),
+		Group:        check.NewGroupChecker(client),
+		//MsgInterface: controller.MsgInterface(),
+		RegisterCenter: client,
+	}
+	s.initPrometheus()
+	msg.RegisterMsgServer(server, s)
+	return nil
 }
 
-func (rpc *rpcChat) initPrometheus() {
-	//sendMsgSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
-	//	Name: "send_msg_success",
-	//	Help: "The number of send msg success",
-	//})
-	//sendMsgFailedCounter = promauto.NewCounter(prometheus.CounterOpts{
-	//	Name: "send_msg_failed",
-	//	Help: "The number of send msg failed",
-	//})
+func (m *msgServer) initPrometheus() {
 	promePkg.NewMsgPullFromRedisSuccessCounter()
 	promePkg.NewMsgPullFromRedisFailedCounter()
 	promePkg.NewMsgPullFromMongoSuccessCounter()
 	promePkg.NewMsgPullFromMongoFailedCounter()
-
 	promePkg.NewSingleChatMsgRecvSuccessCounter()
 	promePkg.NewGroupChatMsgRecvSuccessCounter()
 	promePkg.NewWorkSuperGroupChatMsgRecvSuccessCounter()
-
 	promePkg.NewSingleChatMsgProcessSuccessCounter()
 	promePkg.NewSingleChatMsgProcessFailedCounter()
 	promePkg.NewGroupChatMsgProcessSuccessCounter()
@@ -78,78 +63,3 @@ func (rpc *rpcChat) initPrometheus() {
 	promePkg.NewWorkSuperGroupChatMsgProcessSuccessCounter()
 	promePkg.NewWorkSuperGroupChatMsgProcessFailedCounter()
 }
-
-func (m *msgServer) Run() {
-	log.Info("", "rpcChat init...")
-	listenIP := ""
-	if config.Config.ListenIP == "" {
-		listenIP = "0.0.0.0"
-	} else {
-		listenIP = config.Config.ListenIP
-	}
-	address := listenIP + ":" + strconv.Itoa(m.rpcPort)
-	listener, err := net.Listen("tcp", address)
-	if err != nil {
-		panic("listening err:" + err.Error() + m.rpcRegisterName)
-	}
-	log.Info("", "listen network success, address ", address)
-	recvSize := 1024 * 1024 * 30
-	sendSize := 1024 * 1024 * 30
-	var grpcOpts = []grpc.ServerOption{
-		grpc.MaxRecvMsgSize(recvSize),
-		grpc.MaxSendMsgSize(sendSize),
-	}
-	if config.Config.Prometheus.Enable {
-		promePkg.NewGrpcRequestCounter()
-		promePkg.NewGrpcRequestFailedCounter()
-		promePkg.NewGrpcRequestSuccessCounter()
-		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
-			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
-			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
-		}...)
-	}
-	srv := grpc.NewServer(grpcOpts...)
-	defer srv.GracefulStop()
-
-	rpcRegisterIP := config.Config.RpcRegisterIP
-	msg.RegisterMsgServer(srv, m)
-	if config.Config.RpcRegisterIP == "" {
-		rpcRegisterIP, err = utils.GetLocalIP()
-		if err != nil {
-			log.Error("", "GetLocalIP failed ", err.Error())
-		}
-	}
-	err = getcdv3.RegisterEtcd(m.etcdSchema, strings.Join(m.etcdAddr, ","), rpcRegisterIP, m.rpcPort, m.rpcRegisterName, 10, "")
-	if err != nil {
-		log.Error("", "register rpcChat to etcd failed ", err.Error())
-		panic(utils.Wrap(err, "register chat module  m to etcd err"))
-	}
-	go m.runCh()
-	m.initPrometheus()
-	err = srv.Serve(listener)
-	if err != nil {
-		log.Error("", "m rpcChat failed ", err.Error())
-		return
-	}
-	log.Info("", "m rpcChat init success")
-}
-
-func (rpc *rpcChat) runCh() {
-	log.NewInfo("", "start del msg chan ")
-	for {
-		select {
-		case msg := <-rpc.delMsgCh:
-			log.NewInfo(msg.OperationID, utils.GetSelfFuncName(), "delmsgch recv new: ", msg)
-			db.DB.DelMsgFromCache(msg.UserID, msg.SeqList, msg.OperationID)
-			unexistSeqList, err := db.DB.DelMsgBySeqList(msg.UserID, msg.SeqList, msg.OperationID)
-			if err != nil {
-				log.NewError(msg.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", msg.UserID, msg.SeqList, msg.OperationID, err.Error())
-				continue
-			}
-			if len(unexistSeqList) > 0 {
-				notification.DeleteMessageNotification(msg.OpUserID, msg.UserID, unexistSeqList, msg.OperationID)
-			}
-		}
-	}
-}

internal/rpc/user/user.go

@@ -4,184 +4,46 @@ import (
 	"Open_IM/internal/common/check"
 	"Open_IM/internal/common/convert"
 	"Open_IM/internal/common/notification"
-	"Open_IM/internal/common/rpcserver"
-	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/db/controller"
 	"Open_IM/pkg/common/db/relation"
 	tablerelation "Open_IM/pkg/common/db/table/relation"
-	"Open_IM/pkg/common/log"
-	prome "Open_IM/pkg/common/prometheus"
 	"Open_IM/pkg/common/tokenverify"
 	"Open_IM/pkg/common/tracelog"
+	discoveryRegistry "Open_IM/pkg/discoveryregistry"
 	"Open_IM/pkg/proto/sdkws"
 	pbuser "Open_IM/pkg/proto/user"
 	"Open_IM/pkg/utils"
 	"context"
-	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
-
+	"github.com/OpenIMSDK/openKeeper"
 	"google.golang.org/grpc"
 )
 
 type userServer struct {
-	*rpcserver.RpcServer
 	controller.UserInterface
-	notification *notification.Check
-	userCheck    *check.UserCheck
+	notification        *notification.Check
+	userCheck           *check.UserCheck
+	ConversationChecker *check.ConversationChecker
+	RegisterCenter      discoveryRegistry.SvcDiscoveryRegistry
 }
 
-func NewUserServer(port int) *userServer {
-	r, err := rpcserver.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImUserName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
+func Start(client *openKeeper.ZkClient, server *grpc.Server) error {
+	mysql, err := relation.NewGormDB()
 	if err != nil {
-		panic(err)
+		return err
 	}
-	//mysql init
-	var mysql relation.Mysql
-	var model relation.UserGorm
-	err = mysql.InitConn().AutoMigrateModel(&model)
-	if err != nil {
-		panic("db init err:" + err.Error())
+	if err := mysql.AutoMigrate(&tablerelation.UserModel{}); err != nil {
+		return err
 	}
-	if mysql.GormConn() != nil {
-		model.DB = mysql.GormConn()
-	} else {
-		panic("db init err:" + "conn is nil")
-	}
-	return &userServer{RpcServer: r, UserInterface: controller.NewUserController(model.DB)}
+	pbuser.RegisterUserServer(server, &userServer{
+		UserInterface:  controller.NewUserController(mysql),
+		notification:   notification.NewCheck(client),
+		userCheck:      check.NewUserCheck(client),
+		RegisterCenter: client,
+	})
+	return nil
 }
 
-func (s *userServer) Run() {
-	operationID := utils.OperationIDGenerator()
-	log.NewInfo(operationID, "rpc user start...")
-	listener, address, err := rpcserver.GetTcpListen(config.Config.ListenIP, s.Port)
-	if err != nil {
-		panic(err)
-	}
-	log.NewInfo(operationID, "listen ok ", address)
-	defer listener.Close()
-	//grpc server
-	var grpcOpts []grpc.ServerOption
-	if config.Config.Prometheus.Enable {
-		prome.NewGrpcRequestCounter()
-		prome.NewGrpcRequestFailedCounter()
-		prome.NewGrpcRequestSuccessCounter()
-		grpcOpts = append(grpcOpts, []grpc.ServerOption{
-			// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
-			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
-			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
-		}...)
-	}
-	srv := grpc.NewServer(grpcOpts...)
-	defer srv.GracefulStop()
-	//Service registers with etcd
-	pbuser.RegisterUserServer(srv, s)
-
-	err = srv.Serve(listener)
-	if err != nil {
-		panic(err)
-	}
-	log.NewInfo(operationID, "rpc  user success")
-}
-
-// ok
-//func (s *userServer) SyncJoinedGroupMemberFaceURL(ctx context.Context, userID string, faceURL string, operationID string, opUserID string) {
-//	members, err := s.GetJoinedGroupMembers(ctx, userID)
-//	if err != nil {
-//		return
-//	}
-//	groupIDs := make([]string, 0)
-//	for _, v := range members {
-//		groupIDs = append(groupIDs, v.GroupID)
-//	}
-//	if s.SetGroupMemberInfo(ctx, "", faceURL, "", 0, groupIDs, userID) != nil {
-//		return
-//	}
-//	for _, v := range groupIDs {
-//		chat.GroupMemberInfoSetNotification(operationID, opUserID, v, userID)
-//	}
-//}
-
-// ok
-//func (s *userServer) SyncJoinedGroupMemberNickname(ctx context.Context, userID string, newNickname, oldNickname string, operationID string, opUserID string) {
-//	members, err := s.GetJoinedGroupMembers(ctx, userID)
-//	if err != nil {
-//		return
-//	}
-//	groupIDs := make([]string, 0)
-//	for _, v := range members {
-//		if v.Nickname == oldNickname {
-//			groupIDs = append(groupIDs, v.GroupID)
-//		}
-//	}
-//	s.SetGroupMemberInfo(ctx, newNickname, "", "", 0, groupIDs, userID)
-//	for _, v := range groupIDs {
-//		chat.GroupMemberInfoSetNotification(operationID, opUserID, v, userID)
-//	}
-//}
-
-// 设置群头像
-//func (s *userServer) SetGroupMemberInfo(ctx context.Context, nickname, faceURL, ex string, roleLevel int32, groupIDs []string, userID string) (err error) {
-//
-//	req := pbgroup.SetGroupMemberInfo{UserID: userID}
-//	if nickname != "" {
-//		req.Nickname = &wrappers.StringValue{Value: nickname}
-//	}
-//	if faceURL != "" {
-//		req.FaceURL = &wrappers.StringValue{Value: faceURL}
-//	}
-//	if ex != "" {
-//		req.Ex = &wrappers.StringValue{Value: ex}
-//	}
-//	if roleLevel != 0 {
-//		req.RoleLevel = &wrappers.Int32Value{Value: roleLevel}
-//	}
-//
-//	setGroupMemberInfoReq := &pbgroup.SetGroupMemberInfoReq{}
-//	for _, v := range groupIDs {
-//		req.GroupID = v
-//		setGroupMemberInfoReq.Members = append(setGroupMemberInfoReq.Members, &req)
-//	}
-//	conn, err := s.RegisterCenter.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
-//	if err != nil {
-//		return err
-//	}
-//	client := group.NewGroupClient(conn)
-//	_, err = client.SetGroupMemberInfo(ctx, setGroupMemberInfoReq)
-//	return
-//}
-
-// 获取加入的群成员信息
-//func (s *userServer) GetJoinedGroupMembers(ctx context.Context, userID string) (members []*sdkws.GroupMemberFullInfo, err error) {
-//	conn, err := s.RegisterCenter.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
-//	if err != nil {
-//		return nil, err
-//	}
-//
-//	client := group.NewGroupClient(conn)
-//	for {
-//		idx := int32(0)
-//		req := pbgroup.GetJoinedGroupListReq{FromUserID: userID, Pagination: &sdkws.RequestPagination{PageNumber: idx, ShowNumber: constant.ShowNumber}}
-//		resp, err := client.GetJoinedGroupList(ctx, &req)
-//		if err != nil {
-//			return nil, err
-//		}
-//		groupIDs := make([]string, 0)
-//
-//		for _, v := range resp.Groups {
-//			groupIDs = append(groupIDs, v.GroupID)
-//		}
-//
-//		client.GetGroupMembersInfo()
-//
-//		if len(resp.Groups) < constant.ShowNumber {
-//			break
-//		}
-//		idx++
-//	}
-//
-//	return
-//}
-
 // ok
 func (s *userServer) GetDesignateUsers(ctx context.Context, req *pbuser.GetDesignateUsersReq) (resp *pbuser.GetDesignateUsersResp, err error) {
 	resp = &pbuser.GetDesignateUsersResp{}

internal/startrpc/start.go

@@ -15,15 +15,13 @@ import (
 	"net"
 )
 
-func start(rpcPort int, rpcRegisterName string, prometheusPort int, rpcFn func(client *openKeeper.ZkClient, server *grpc.Server) error, options []grpc.ServerOption) error {
-	flagRpcPort := flag.Int("port", rpcPort, "get RpcGroupPort from cmd,default 16000 as port")
-	flagPrometheusPort := flag.Int("prometheus_port", prometheusPort, "groupPrometheusPort default listen port")
+func start(rpcPorts []int, rpcRegisterName string, prometheusPorts []int, rpcFn func(client *openKeeper.ZkClient, server *grpc.Server) error, options []grpc.ServerOption) error {
+	flagRpcPort := flag.Int("port", rpcPorts[0], "get RpcGroupPort from cmd,default 16000 as port")
+	flagPrometheusPort := flag.Int("prometheus_port", prometheusPorts[0], "groupPrometheusPort default listen port")
 	flag.Parse()
-	rpcPort = *flagRpcPort
-	prometheusPort = *flagPrometheusPort
-	fmt.Println("start group rpc server, port: ", rpcPort, ", OpenIM version: ", constant.CurrentVersion)
+	fmt.Println("start group rpc server, port: ", *flagRpcPort, ", OpenIM version: ", constant.CurrentVersion)
 	log.NewPrivateLog(constant.LogFileName)
-	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, rpcPort))
+	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, *flagRpcPort))
 	if err != nil {
 		return err
 	}
@@ -50,12 +48,12 @@ func start(rpcPort int, rpcRegisterName string, prometheusPort int, rpcFn func(c
 	}
 	srv := grpc.NewServer(options...)
 	defer srv.GracefulStop()
-	err = zkClient.Register(rpcRegisterName, registerIP, rpcPort)
+	err = zkClient.Register(rpcRegisterName, registerIP, *flagRpcPort)
 	if err != nil {
 		return err
 	}
 	if config.Config.Prometheus.Enable {
-		err := promePkg.StartPromeSrv(prometheusPort)
+		err := promePkg.StartPromeSrv(*flagPrometheusPort)
 		if err != nil {
 			return err
 		}
@@ -63,7 +61,7 @@ func start(rpcPort int, rpcRegisterName string, prometheusPort int, rpcFn func(c
 	return rpcFn(zkClient, srv)
 }
 
-func Start(rpcPort int, rpcRegisterName string, prometheusPort int, rpcFn func(client *openKeeper.ZkClient, server *grpc.Server) error, options ...grpc.ServerOption) {
-	err := start(rpcPort, rpcRegisterName, prometheusPort, rpcFn, options)
+func Start(rpcPorts []int, rpcRegisterName string, prometheusPorts []int, rpcFn func(client *openKeeper.ZkClient, server *grpc.Server) error, options ...grpc.ServerOption) {
+	err := start(rpcPorts, rpcRegisterName, prometheusPorts, rpcFn, options)
 	fmt.Println("end", err)
 }