Netty optimization

pkg/client/client.go

@@ -62,15 +62,17 @@ func initTmClient(cfg *Config) {
 	})
 }
 
-// initRemoting init rpc client
+// initRemoting init remoting
 func initRemoting(cfg *Config) {
-	getty.InitRpcClient(&cfg.GettyConfig, &remoteConfig.SeataConfig{
+	SeataConfig := remoteConfig.SeataConfig{
 		ApplicationID:        cfg.ApplicationID,
 		TxServiceGroup:       cfg.TxServiceGroup,
 		ServiceVgroupMapping: cfg.ServiceConfig.VgroupMapping,
 		ServiceGrouplist:     cfg.ServiceConfig.Grouplist,
 		LoadBalanceType:      cfg.GettyConfig.LoadBalanceType,
-	})
+	}
+
+	getty.InitGetty(&cfg.GettyConfig, &SeataConfig)
 }
 
 // InitRmClient init client rm client

pkg/remoting/config/config.go

@@ -84,7 +84,7 @@ type SeataConfig struct {
 	LoadBalanceType      string
 }
 
-func IniConfig(seataConf *SeataConfig) {
+func InitConfig(seataConf *SeataConfig) {
 	seataConfig = seataConf
 }
 

pkg/remoting/getty/getty_init.go

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package getty
+
+import (
+	"seata.apache.org/seata-go/pkg/protocol/codec"
+	"seata.apache.org/seata-go/pkg/remoting/config"
+)
+
+func InitGetty(gettyConfig *config.Config, seataConfig *config.SeataConfig) {
+	config.InitConfig(seataConfig)
+	codec.Init()
+	initSessionManager(gettyConfig, seataConfig)
+}

pkg/remoting/getty/listener.go

@@ -38,22 +38,20 @@ var (
 )
 
 type gettyClientHandler struct {
-	idGenerator    *atomic.Uint32
-	msgFutures     *sync.Map
-	mergeMsgMap    *sync.Map
-	sessionManager *SessionManager
-	processorMap   map[message.MessageType]processor.RemotingProcessor
+	idGenerator  *atomic.Uint32
+	msgFutures   *sync.Map
+	mergeMsgMap  *sync.Map
+	processorMap map[message.MessageType]processor.RemotingProcessor
 }
 
 func GetGettyClientHandlerInstance() *gettyClientHandler {
 	if clientHandler == nil {
 		onceClientHandler.Do(func() {
 			clientHandler = &gettyClientHandler{
-				idGenerator:    &atomic.Uint32{},
-				msgFutures:     &sync.Map{},
-				mergeMsgMap:    &sync.Map{},
-				sessionManager: sessionManager,
-				processorMap:   make(map[message.MessageType]processor.RemotingProcessor, 0),
+				idGenerator:  &atomic.Uint32{},
+				msgFutures:   &sync.Map{},
+				mergeMsgMap:  &sync.Map{},
+				processorMap: make(map[message.MessageType]processor.RemotingProcessor, 0),
 			}
 		})
 	}
@@ -62,7 +60,7 @@ func GetGettyClientHandlerInstance() *gettyClientHandler {
 
 func (g *gettyClientHandler) OnOpen(session getty.Session) error {
 	log.Infof("Open new getty session ")
-	g.sessionManager.registerSession(session)
+	sessionManager.registerSession(session)
 	conf := config.GetSeataConfig()
 	go func() {
 		request := message.RegisterTMRequest{AbstractIdentifyRequest: message.AbstractIdentifyRequest{
@@ -73,7 +71,7 @@ func (g *gettyClientHandler) OnOpen(session getty.Session) error {
 		err := GetGettyRemotingClient().SendAsyncRequest(request)
 		if err != nil {
 			log.Errorf("OnOpen error: {%#v}", err.Error())
-			g.sessionManager.releaseSession(session)
+			sessionManager.releaseSession(session)
 			return
 		}
 	}()
@@ -83,12 +81,12 @@ func (g *gettyClientHandler) OnOpen(session getty.Session) error {
 
 func (g *gettyClientHandler) OnError(session getty.Session, err error) {
 	log.Infof("session{%s} got error{%v}, will be closed.", session.Stat(), err)
-	g.sessionManager.releaseSession(session)
+	sessionManager.releaseSession(session)
 }
 
 func (g *gettyClientHandler) OnClose(session getty.Session) {
 	log.Infof("session{%s} is closing......", session.Stat())
-	g.sessionManager.releaseSession(session)
+	sessionManager.releaseSession(session)
 }
 
 func (g *gettyClientHandler) OnMessage(session getty.Session, pkg interface{}) {
@@ -118,7 +116,7 @@ func (g *gettyClientHandler) OnCron(session getty.Session) {
 	err := g.transferHeartBeat(session, message.HeartBeatMessagePing)
 	if err != nil {
 		log.Errorf("failed to send heart beat: {%#v}", err.Error())
-		g.sessionManager.releaseSession(session)
+		sessionManager.releaseSession(session)
 	}
 }