fix: getty session auto close bug

pkg/common/log/logging.go

@@ -104,7 +104,15 @@ var (
 	log       Logger
 	zapLogger *zap.Logger
 
-	zapLoggerConfig        = zap.NewDevelopmentConfig()
+	zapLoggerConfig = zap.Config{
+		// todo read level from config
+		Level:            zap.NewAtomicLevelAt(zap.InfoLevel),
+		Development:      true,
+		Encoding:         "console",
+		EncoderConfig:    zap.NewDevelopmentEncoderConfig(),
+		OutputPaths:      []string{"stderr"},
+		ErrorOutputPaths: []string{"stderr"},
+	}
 	zapLoggerEncoderConfig = zapcore.EncoderConfig{
 		TimeKey:        "time",
 		LevelKey:       "level",

pkg/config/getty_config.go

@@ -38,8 +38,8 @@ type GettyConfig struct {
 // GetDefaultGettyConfig ...
 func GetDefaultGettyConfig() GettyConfig {
 	return GettyConfig{
-		ReconnectInterval: 1,
-		ConnectionNum:     20,
+		ReconnectInterval: 0,
+		ConnectionNum:     1,
 		HeartbeatPeriod:   10 * time.Second,
 		GettySessionParam: GettySessionParam{
 			CompressEncoding: false,
@@ -51,7 +51,7 @@ func GetDefaultGettyConfig() GettyConfig {
 			TCPReadTimeout:   time.Second,
 			TCPWriteTimeout:  5 * time.Second,
 			WaitTimeout:      time.Second,
-			CronPeriod:       5 * time.Second,
+			CronPeriod:       time.Second,
 			MaxMsgLen:        4096,
 			SessionName:      "rpc_client",
 		},

pkg/protocol/codec/branch_commit_response_codec.go

@@ -39,7 +39,7 @@ func (g *BranchCommitResponseCodec) Decode(in []byte) interface{} {
 
 	data.ResultCode = message.ResultCode(bytes.ReadByte(buf))
 	if data.ResultCode == message.ResultCodeFailed {
-		data.Msg = bytes.ReadString16Length(buf)
+		data.Msg = bytes.ReadString8Length(buf)
 	}
 	data.TransactionExceptionCode = serror.TransactionExceptionCode(bytes.ReadByte(buf))
 	data.Xid = bytes.ReadString16Length(buf)
@@ -56,10 +56,10 @@ func (g *BranchCommitResponseCodec) Encode(in interface{}) []byte {
 	buf.WriteByte(byte(data.ResultCode))
 	if data.ResultCode == message.ResultCodeFailed {
 		msg := data.Msg
-		if len(data.Msg) > math.MaxInt16 {
-			msg = data.Msg[:math.MaxInt16]
+		if len(data.Msg) > math.MaxInt8 {
+			msg = data.Msg[:math.MaxInt8]
 		}
-		bytes.WriteString16Length(msg, buf)
+		bytes.WriteString8Length(msg, buf)
 	}
 	buf.WriteByte(byte(data.TransactionExceptionCode))
 	bytes.WriteString16Length(data.Xid, buf)

pkg/protocol/codec/branch_rollback_response_codec.go

@@ -39,7 +39,7 @@ func (g *BranchRollbackResponseCodec) Decode(in []byte) interface{} {
 
 	data.ResultCode = message.ResultCode(bytes.ReadByte(buf))
 	if data.ResultCode == message.ResultCodeFailed {
-		data.Msg = bytes.ReadString16Length(buf)
+		data.Msg = bytes.ReadString8Length(buf)
 	}
 	data.TransactionExceptionCode = serror.TransactionExceptionCode(bytes.ReadByte(buf))
 	data.Xid = bytes.ReadString16Length(buf)
@@ -59,7 +59,7 @@ func (g *BranchRollbackResponseCodec) Encode(in interface{}) []byte {
 		if len(data.Msg) > math.MaxInt16 {
 			msg = data.Msg[:math.MaxInt16]
 		}
-		bytes.WriteString16Length(msg, buf)
+		bytes.WriteString8Length(msg, buf)
 	}
 	buf.WriteByte(byte(data.TransactionExceptionCode))
 	bytes.WriteString16Length(data.Xid, buf)

pkg/remoting/getty/getty_client.go

@@ -19,8 +19,10 @@ package getty
 
 import (
 	"sync"
-	"time"
 
+	gxtime "github.com/dubbogo/gost/time"
+	"github.com/pkg/errors"
+	"github.com/seata/seata-go/pkg/common/log"
 	"github.com/seata/seata-go/pkg/protocol/codec"
 	"github.com/seata/seata-go/pkg/protocol/message"
 	"go.uber.org/atomic"
@@ -60,18 +62,18 @@ func (client *GettyRemotingClient) SendAsyncRequest(msg interface{}) error {
 		Compressor: 0,
 		Body:       msg,
 	}
-	return GetGettyRemotingInstance().SendASync(rpcMessage)
+	return GetGettyRemotingInstance().SendASync(rpcMessage, nil, client.asyncCallback)
 }
 
-func (client *GettyRemotingClient) SendAsyncResponse(msg interface{}) error {
+func (client *GettyRemotingClient) SendAsyncResponse(msgID int32, msg interface{}) error {
 	rpcMessage := message.RpcMessage{
-		ID:         int32(client.idGenerator.Inc()),
+		ID:         msgID,
 		Type:       message.GettyRequestType_Response,
 		Codec:      byte(codec.CodecTypeSeata),
 		Compressor: 0,
 		Body:       msg,
 	}
-	return GetGettyRemotingInstance().SendASync(rpcMessage)
+	return GetGettyRemotingInstance().SendASync(rpcMessage, nil, nil)
 }
 
 func (client *GettyRemotingClient) SendSyncRequest(msg interface{}) (interface{}, error) {
@@ -82,16 +84,21 @@ func (client *GettyRemotingClient) SendSyncRequest(msg interface{}) (interface{}
 		Compressor: 0,
 		Body:       msg,
 	}
-	return GetGettyRemotingInstance().SendSync(rpcMessage)
+	return GetGettyRemotingInstance().SendSync(rpcMessage, nil, client.syncCallback)
 }
 
-func (client *GettyRemotingClient) SendSyncRequestWithTimeout(msg interface{}, timeout time.Duration) (interface{}, error) {
-	rpcMessage := message.RpcMessage{
-		ID:         int32(client.idGenerator.Inc()),
-		Type:       message.GettyRequestType_RequestSync,
-		Codec:      byte(codec.CodecTypeSeata),
-		Compressor: 0,
-		Body:       msg,
+func (g *GettyRemotingClient) asyncCallback(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error) {
+	go g.asyncCallback(reqMsg, respMsg)
+	return nil, nil
+}
+
+func (g *GettyRemotingClient) syncCallback(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error) {
+	select {
+	case <-gxtime.GetDefaultTimerWheel().After(RPC_REQUEST_TIMEOUT):
+		GetGettyRemotingInstance().RemoveMergedMessageFuture(reqMsg.ID)
+		log.Errorf("wait resp timeout: %#v", reqMsg)
+		return nil, errors.Errorf("wait response timeout, request: %#v", reqMsg)
+	case <-respMsg.Done:
+		return respMsg.Response, respMsg.Err
 	}
-	return GetGettyRemotingInstance().SendSyncWithTimeout(rpcMessage, timeout)
 }

pkg/remoting/getty/getty_remoting.go

@@ -23,25 +23,26 @@ import (
 
 	getty "github.com/apache/dubbo-getty"
 
-	gxtime "github.com/dubbogo/gost/time"
-	"github.com/pkg/errors"
 	"github.com/seata/seata-go/pkg/common/log"
 	"github.com/seata/seata-go/pkg/protocol/message"
 )
 
 const (
-	RPC_REQUEST_TIMEOUT = 5 * time.Second
+	RPC_REQUEST_TIMEOUT = 2 * time.Second
 )
 
 var (
 	gettyRemoting     *GettyRemoting
 	onceGettyRemoting = &sync.Once{}
 )
 
-type GettyRemoting struct {
-	futures     *sync.Map
-	mergeMsgMap *sync.Map
-}
+type (
+	callbackMethod func(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error)
+	GettyRemoting  struct {
+		futures     *sync.Map
+		mergeMsgMap *sync.Map
+	}
+)
 
 func GetGettyRemotingInstance() *GettyRemoting {
 	if gettyRemoting == nil {
@@ -55,67 +56,44 @@ func GetGettyRemotingInstance() *GettyRemoting {
 	return gettyRemoting
 }
 
-func (client *GettyRemoting) SendSync(msg message.RpcMessage) (interface{}, error) {
-	ss := sessionManager.AcquireGettySession()
-	return client.sendAsync(ss, msg, RPC_REQUEST_TIMEOUT)
-}
-
-func (client *GettyRemoting) SendSyncWithTimeout(msg message.RpcMessage, timeout time.Duration) (interface{}, error) {
-	ss := sessionManager.AcquireGettySession()
-	return client.sendAsync(ss, msg, timeout)
+func (g *GettyRemoting) SendSync(msg message.RpcMessage, s getty.Session, callback callbackMethod) (interface{}, error) {
+	if s == nil {
+		s = sessionManager.selectSession()
+	}
+	return g.sendAsync(s, msg, callback)
 }
 
-func (client *GettyRemoting) SendASync(msg message.RpcMessage) error {
-	ss := sessionManager.AcquireGettySession()
-	_, err := client.sendAsync(ss, msg, 0*time.Second)
+func (g *GettyRemoting) SendASync(msg message.RpcMessage, s getty.Session, callback callbackMethod) error {
+	if s == nil {
+		s = sessionManager.selectSession()
+	}
+	_, err := g.sendAsync(s, msg, callback)
 	return err
 }
 
-func (client *GettyRemoting) sendAsync(session getty.Session, msg message.RpcMessage, timeout time.Duration) (interface{}, error) {
-	log.Infof("send async message: {%#v}", msg)
+func (g *GettyRemoting) sendAsync(session getty.Session, msg message.RpcMessage, callback callbackMethod) (interface{}, error) {
+	if _, ok := msg.Body.(message.HeartBeatMessage); ok {
+		log.Debug("send async message: {%#v}", msg)
+	} else {
+		log.Infof("send async message: {%#v}", msg)
+	}
 	var err error
 	if session == nil || session.IsClosed() {
 		log.Warn("sendAsyncRequestWithResponse nothing, caused by null channel.")
 		return nil, err
 	}
 	resp := message.NewMessageFuture(msg)
-	client.futures.Store(msg.ID, resp)
+	g.futures.Store(msg.ID, resp)
 	_, _, err = session.WritePkg(msg, time.Duration(0))
 	if err != nil {
-		client.futures.Delete(msg.ID)
+		g.futures.Delete(msg.ID)
 		log.Errorf("send message: %#v, session: %s", msg, session.Stat())
 		return nil, err
 	}
-
-	log.Debugf("send message: %#v, session: %s", msg, session.Stat())
-
-	actualTimeOut := timeout
-	if timeout <= time.Duration(0) {
-		// todo timeoue use config
-		actualTimeOut = time.Duration(2000)
-	}
-
-	wait := func() (interface{}, error) {
-		select {
-		case <-gxtime.GetDefaultTimerWheel().After(actualTimeOut):
-			client.futures.Delete(msg.ID)
-			if session != nil {
-				return nil, errors.Errorf("wait response timeout, ip: %s, request: %#v", session.RemoteAddr(), msg)
-			} else {
-				return nil, errors.Errorf("wait response timeout and session is nil, request: %#v", msg)
-			}
-		case <-resp.Done:
-			err = resp.Err
-			return resp.Response, err
-		}
-	}
-
-	if timeout > time.Duration(0) {
-		return wait()
-	} else {
-		go wait()
+	if callback != nil {
+		return callback(msg, resp)
 	}
-	return nil, err
+	return nil, nil
 }
 
 func (client *GettyRemoting) GetMessageFuture(msgID int32) *message.MessageFuture {

pkg/remoting/getty/listener.go

@@ -24,6 +24,7 @@ import (
 	getty "github.com/apache/dubbo-getty"
 	"github.com/seata/seata-go/pkg/common/log"
 	"github.com/seata/seata-go/pkg/config"
+	"github.com/seata/seata-go/pkg/protocol/codec"
 	"github.com/seata/seata-go/pkg/protocol/message"
 	"github.com/seata/seata-go/pkg/remoting/processor"
 	"go.uber.org/atomic"
@@ -39,7 +40,8 @@ type gettyClientHandler struct {
 	idGenerator    *atomic.Uint32
 	msgFutures     *sync.Map
 	mergeMsgMap    *sync.Map
-	processorTable map[message.MessageType]processor.RemotingProcessor
+	sessionManager *SessionManager
+	processorMap   map[message.MessageType]processor.RemotingProcessor
 }
 
 func GetGettyClientHandlerInstance() *gettyClientHandler {
@@ -50,70 +52,84 @@ func GetGettyClientHandlerInstance() *gettyClientHandler {
 				idGenerator:    &atomic.Uint32{},
 				msgFutures:     &sync.Map{},
 				mergeMsgMap:    &sync.Map{},
-				processorTable: make(map[message.MessageType]processor.RemotingProcessor, 0),
+				sessionManager: sessionManager,
+				processorMap:   make(map[message.MessageType]processor.RemotingProcessor, 0),
 			}
 		})
 	}
 	return clientHandler
 }
 
-func (client *gettyClientHandler) OnOpen(session getty.Session) error {
-	sessionManager.RegisterGettySession(session)
+func (g *gettyClientHandler) OnOpen(session getty.Session) error {
+	log.Infof("Open new getty session ")
+	g.sessionManager.registerSession(session)
 	go func() {
 		request := message.RegisterTMRequest{AbstractIdentifyRequest: message.AbstractIdentifyRequest{
-			Version:                 client.conf.SeataVersion,
-			ApplicationId:           client.conf.ApplicationID,
-			TransactionServiceGroup: client.conf.TransactionServiceGroup,
+			Version:                 g.conf.SeataVersion,
+			ApplicationId:           g.conf.ApplicationID,
+			TransactionServiceGroup: g.conf.TransactionServiceGroup,
 		}}
 		err := GetGettyRemotingClient().SendAsyncRequest(request)
-		//client.sendAsyncRequestWithResponse(session, request, RPC_REQUEST_TIMEOUT)
 		if err != nil {
 			log.Errorf("OnOpen error: {%#v}", err.Error())
-			sessionManager.ReleaseGettySession(session)
+			g.sessionManager.releaseSession(session)
 			return
 		}
-		//todo
-		//client.GettySessionOnOpenChannel <- session.RemoteAddr()
 	}()
+
 	return nil
 }
 
-func (client *gettyClientHandler) OnError(session getty.Session, err error) {
-	log.Infof("OnError session{%s} got error{%v}, will be closed.", session.Stat(), err)
-	sessionManager.ReleaseGettySession(session)
+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)
 }
 
-func (client *gettyClientHandler) OnClose(session getty.Session) {
-	log.Infof("OnClose session{%s} is closing......", session.Stat())
-	sessionManager.ReleaseGettySession(session)
+func (g *gettyClientHandler) OnClose(session getty.Session) {
+	log.Infof("session{%s} is closing......", session.Stat())
+	g.sessionManager.releaseSession(session)
 }
 
-func (client *gettyClientHandler) OnMessage(session getty.Session, pkg interface{}) {
+func (g *gettyClientHandler) OnMessage(session getty.Session, pkg interface{}) {
 	ctx := context.Background()
-	log.Debugf("received message: {%#v}", pkg)
+	log.Debug("received message: {%#v}", pkg)
+
 	rpcMessage, ok := pkg.(message.RpcMessage)
 	if !ok {
 		log.Errorf("received message is not protocol.RpcMessage. pkg: %#v", pkg)
 		return
 	}
+
 	if mm, ok := rpcMessage.Body.(message.MessageTypeAware); ok {
-		processor := client.processorTable[mm.GetTypeCode()]
+		processor := g.processorMap[mm.GetTypeCode()]
 		if processor != nil {
 			processor.Process(ctx, rpcMessage)
 		} else {
-			log.Errorf("This message type [%v] has no processor.", mm.GetTypeCode())
+			log.Errorf("This message type %v has no processor.", mm.GetTypeCode())
 		}
 	} else {
 		log.Errorf("This rpcMessage body %#v is not MessageTypeAware type.", rpcMessage.Body)
 	}
 }
 
-func (client *gettyClientHandler) OnCron(session getty.Session) {
-	//GetGettyRemotingClient().SendAsyncRequest(message.HeartBeatMessagePing)
+func (g *gettyClientHandler) OnCron(session getty.Session) {
+	log.Debug("session{%s} Oncron executing", session.Stat())
+	g.transferBeatHeart(session, message.HeartBeatMessagePing)
+}
+
+func (g *gettyClientHandler) transferBeatHeart(session getty.Session, msg message.HeartBeatMessage) {
+	rpcMessage := message.RpcMessage{
+		ID:         int32(g.idGenerator.Inc()),
+		Type:       message.GettyRequestType_HeartbeatRequest,
+		Codec:      byte(codec.CodecTypeSeata),
+		Compressor: 0,
+		Body:       msg,
+	}
+	GetGettyRemotingInstance().SendASync(rpcMessage, session, nil)
 }
 
-func (client *gettyClientHandler) RegisterProcessor(msgType message.MessageType, processor processor.RemotingProcessor) {
+func (g *gettyClientHandler) RegisterProcessor(msgType message.MessageType, processor processor.RemotingProcessor) {
 	if nil != processor {
-		client.processorTable[msgType] = processor
+		g.processorMap[msgType] = processor
 	}
 }