refactor: Refactor rpc call && auto gen rpc_call code (#2969)

* refactor: rpcclient

* chore: err

* fix: err

* fix: err

* fix: err

* feat: change api

internal/msgtransfer/init.go

@@ -25,6 +25,8 @@ import (
 	"strconv"
 	"syscall"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
+	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/utils/jsonutil"
 	"github.com/openimsdk/tools/utils/network"
@@ -39,9 +41,7 @@ import (
 
 	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	discRegister "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mw"
@@ -93,6 +93,9 @@ func Start(ctx context.Context, index int, config *Config) error {
 	}
 	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	if err = rpcclient.InitRpcCaller(client, config.Discovery.RpcService); err != nil {
+		return err
+	}
 
 	msgModel := redis.NewMsgCache(rdb)
 	msgDocModel, err := mgo.NewMsgMongo(mgocli.GetDB())
@@ -113,9 +116,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	conversationRpcClient := rpcclient.NewConversationRpcClient(client, config.Discovery.RpcService.Conversation)
-	groupRpcClient := rpcclient.NewGroupRpcClient(client, config.Discovery.RpcService.Group)
-	historyCH, err := NewOnlineHistoryRedisConsumerHandler(&config.KafkaConfig, msgTransferDatabase, &conversationRpcClient, &groupRpcClient)
+	historyCH, err := NewOnlineHistoryRedisConsumerHandler(&config.KafkaConfig, msgTransferDatabase)
 	if err != nil {
 		return err
 	}
@@ -129,10 +130,10 @@ func Start(ctx context.Context, index int, config *Config) error {
 		historyMongoCH: historyMongoCH,
 		runTimeEnv:     runTimeEnv,
 	}
-	return msgTransfer.Start(index, config)
+	return msgTransfer.Start(index, config, client)
 }
 
-func (m *MsgTransfer) Start(index int, config *Config) error {
+func (m *MsgTransfer) Start(index int, config *Config, client discovery.SvcDiscoveryRegistry) error {
 	m.ctx, m.cancel = context.WithCancel(context.Background())
 	var (
 		netDone = make(chan struct{}, 1)
@@ -147,11 +148,6 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
 		return err
 	}
 
-	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, m.runTimeEnv)
-	if err != nil {
-		return errs.WrapMsg(err, "failed to register discovery service")
-	}
-
 	registerIP, err := network.GetRpcRegisterIP("")
 	if err != nil {
 		return err

internal/msgtransfer/online_history_msg_handler.go

@@ -23,16 +23,17 @@ import (
 	"sync"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-
 	"github.com/IBM/sarama"
 	"github.com/go-redis/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/open-im-server/v3/pkg/tools/batcher"
 	"github.com/openimsdk/protocol/constant"
+	pbconv "github.com/openimsdk/protocol/conversation"
+	"github.com/openimsdk/protocol/group"
+	"github.com/openimsdk/protocol/rpccall"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
@@ -69,14 +70,11 @@ type OnlineHistoryRedisConsumerHandler struct {
 	redisMessageBatches *batcher.Batcher[sarama.ConsumerMessage]
 
 	msgTransferDatabase         controller.MsgTransferDatabase
-	conversationRpcClient       *rpcclient.ConversationRpcClient
-	groupRpcClient              *rpcclient.GroupRpcClient
 	conversationUserHasReadChan chan *userHasReadSeq
 	wg                          sync.WaitGroup
 }
 
-func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase,
-	conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) (*OnlineHistoryRedisConsumerHandler, error) {
+func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase) (*OnlineHistoryRedisConsumerHandler, error) {
 	historyConsumerGroup, err := kafka.NewMConsumerGroup(kafkaConf.Build(), kafkaConf.ToRedisGroupID, []string{kafkaConf.ToRedisTopic}, false)
 	if err != nil {
 		return nil, err
@@ -103,8 +101,6 @@ func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database cont
 	}
 	b.Do = och.do
 	och.redisMessageBatches = b
-	och.conversationRpcClient = conversationRpcClient
-	och.groupRpcClient = groupRpcClient
 	och.historyConsumerGroup = historyConsumerGroup
 
 	return &och, err
@@ -285,22 +281,32 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key
 			case constant.ReadGroupChatType:
 				log.ZDebug(ctx, "group chat first create conversation", "conversationID",
 					conversationID)
-				userIDs, err := och.groupRpcClient.GetGroupMemberIDs(ctx, msg.GroupID)
+
+				userIDs, err := rpccall.ExtractField(ctx, group.GetGroupMemberUserIDsCaller.Invoke,
+					&group.GetGroupMemberUserIDsReq{
+						GroupID: msg.GroupID,
+					}, (*group.GetGroupMemberUserIDsResp).GetUserIDs)
 				if err != nil {
 					log.ZWarn(ctx, "get group member ids error", err, "conversationID",
 						conversationID)
 				} else {
 					log.ZInfo(ctx, "GetGroupMemberIDs end")
 
-					if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx,
-						msg.GroupID, userIDs); err != nil {
+					if err := pbconv.CreateGroupChatConversationsCaller.Execute(ctx, &pbconv.CreateGroupChatConversationsReq{
+						UserIDs: userIDs,
+						GroupID: msg.GroupID,
+					}); err != nil {
 						log.ZWarn(ctx, "single chat first create conversation error", err,
 							"conversationID", conversationID)
 					}
 				}
 			case constant.SingleChatType, constant.NotificationChatType:
-				if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, msg.RecvID,
-					msg.SendID, conversationID, msg.SessionType); err != nil {
+				if err := pbconv.CreateSingleChatConversationsCaller.Execute(ctx, &pbconv.CreateSingleChatConversationsReq{
+					RecvID:           msg.RecvID,
+					SendID:           msg.SendID,
+					ConversationID:   conversationID,
+					ConversationType: msg.SessionType,
+				}); err != nil {
 					log.ZWarn(ctx, "single chat or notification first create conversation error", err,
 						"conversationID", conversationID, "sessionType", msg.SessionType)
 				}