Feat add two phase (#122)

* add two phase

* support seata dubbo

go.mod

@@ -3,20 +3,17 @@ module github.com/seata/seata-go
 go 1.16
 
 require (
+	dubbo.apache.org/dubbo-go/v3 v3.0.2-0.20220508105316-b27ec53b7bab
 	github.com/BurntSushi/toml v1.1.0 // indirect
 	github.com/apache/dubbo-getty v1.4.8
-	github.com/dubbogo/gost v1.12.3
-	github.com/golang/snappy v0.0.4 // indirect
-	github.com/kr/pretty v0.3.0 // indirect
+	github.com/apache/dubbo-go-hessian2 v1.11.0
+	github.com/dubbogo/gost v1.12.5
 	github.com/natefinch/lumberjack v2.0.0+incompatible
 	github.com/pkg/errors v0.9.1
-	github.com/rogpeppe/go-internal v1.8.0 // indirect
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.7.1
 	go.uber.org/atomic v1.9.0
-	go.uber.org/zap v1.19.1
+	go.uber.org/zap v1.21.0
 	golang.org/x/tools v0.1.11 // indirect
-	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
-	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	vimagination.zapto.org/byteio v0.0.0-20200222190125-d27cba0f0b10
 	vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d // indirect
 )

pkg/common/constants.go

@@ -21,4 +21,12 @@ const (
 	StartTime     = "action-start-time"
 	HostName      = "host-name"
 	ActionContext = "actionContext"
+
+	SeataXidKey    = "SEATA_XID"
+	XidKey         = "TX_XID"
+	MdcXidKey      = "X-TX-XID"
+	MdcBranchIDKey = "X-TX-BRANCH-ID"
+	BranchTypeKey  = "TX_BRANCH_TYPE"
+	GlobalLockKey  = "TX_LOCK"
+	SeataFilterKey = "seataDubboFilter"
 )

pkg/common/log/logging.go

@@ -22,7 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"time"
-	
+
 	getty "github.com/apache/dubbo-getty"
 	"github.com/natefinch/lumberjack"
 	"go.uber.org/zap"

pkg/common/types/types.go

@@ -0,0 +1,49 @@
+/*
+ * 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 types
+
+import "reflect"
+
+type ReferencedService interface {
+	Reference() string
+}
+
+// GetReference return the reference id of the service.
+// If the service implemented the ReferencedService interface,
+// it will call the Reference method. If not, it will
+// return the struct name as the reference id.
+func GetReference(service interface{}) string {
+	if s, ok := service.(ReferencedService); ok {
+		return s.Reference()
+	}
+	ref := ""
+	sType := reflect.TypeOf(service)
+	kind := sType.Kind()
+	switch kind {
+	case reflect.Struct:
+		ref = sType.Name()
+	case reflect.Ptr:
+		sName := sType.Elem().Name()
+		if sName != "" {
+			ref = sName
+		} else {
+			ref = sType.Elem().Field(0).Name
+		}
+	}
+	return ref
+}

pkg/config/getty_config.go

@@ -38,8 +38,8 @@ type GettyConfig struct {
 // GetDefaultGettyConfig ...
 func GetDefaultGettyConfig() GettyConfig {
 	return GettyConfig{
-		ReconnectInterval: 0,
-		ConnectionNum:     2,
+		ReconnectInterval: 1,
+		ConnectionNum:     20,
 		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:       time.Second,
+			CronPeriod:       5 * time.Second,
 			MaxMsgLen:        4096,
 			SessionName:      "rpc_client",
 		},

pkg/integration/dubbo/dubbo_transaction_filter.go

@@ -0,0 +1,96 @@
+/*
+ * 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 dubbo
+
+import (
+	"context"
+	"strings"
+	"sync"
+
+	"dubbo.apache.org/dubbo-go/v3/filter"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+	"github.com/seata/seata-go/pkg/common"
+	"github.com/seata/seata-go/pkg/common/log"
+	"github.com/seata/seata-go/pkg/tm"
+)
+
+var (
+	seataFilter *dubboTransactionFilter
+	once        sync.Once
+)
+
+type Filter interface {
+}
+
+type dubboTransactionFilter struct {
+}
+
+func GetDubboTransactionFilter() filter.Filter {
+	if seataFilter == nil {
+		once.Do(func() {
+			seataFilter = &dubboTransactionFilter{}
+		})
+	}
+	return seataFilter
+}
+
+func (d *dubboTransactionFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	xid := tm.GetXID(ctx)
+	rpcXid := d.getRpcXid(invocation)
+	log.Infof("xid in context is %s, xid in RpcContextis %s", xid, rpcXid)
+
+	if xid != "" {
+		// dubbo go
+		invocation.SetAttachment(common.SeataXidKey, xid)
+		// dubbo java
+		invocation.SetAttachment(common.XidKey, xid)
+	} else if rpcXid != xid {
+		ctx = tm.InitSeataContext(ctx)
+		tm.SetXIDCopy(ctx, rpcXid)
+	}
+	return invoker.Invoke(ctx, invocation)
+	// todo why should unbind xid???
+}
+
+func (*dubboTransactionFilter) OnResponse(ctx context.Context, result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	return result
+}
+
+func (d *dubboTransactionFilter) getRpcXid(invocation protocol.Invocation) string {
+	rpcXid := d.getDubboGoRpcXid(invocation)
+	if rpcXid == "" {
+		rpcXid = d.getDubboJavaRpcXid(invocation)
+	}
+	return rpcXid
+}
+
+func (*dubboTransactionFilter) getDubboGoRpcXid(invocation protocol.Invocation) string {
+	rpcXid := invocation.GetAttachmentWithDefaultValue(common.SeataXidKey, "")
+	if rpcXid == "" {
+		rpcXid = invocation.GetAttachmentWithDefaultValue(strings.ToLower(common.SeataXidKey), "")
+	}
+	return rpcXid
+}
+
+func (*dubboTransactionFilter) getDubboJavaRpcXid(invocation protocol.Invocation) string {
+	rpcXid := invocation.GetAttachmentWithDefaultValue(common.XidKey, "")
+	if rpcXid == "" {
+		rpcXid = invocation.GetAttachmentWithDefaultValue(strings.ToLower(common.XidKey), "")
+	}
+	return rpcXid
+}

pkg/integration/integration.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 integration
+
+import (
+	"dubbo.apache.org/dubbo-go/v3/common/extension"
+
+	"github.com/seata/seata-go/pkg/common"
+	"github.com/seata/seata-go/pkg/integration/dubbo"
+)
+
+func UseDubbo() {
+	extension.SetFilter(common.SeataFilterKey, dubbo.GetDubboTransactionFilter)
+}

pkg/remoting/getty/getty_remoting.go

@@ -30,7 +30,7 @@ import (
 )
 
 const (
-	RPC_REQUEST_TIMEOUT = 30 * time.Second
+	RPC_REQUEST_TIMEOUT = 5 * time.Second
 )
 
 var (

pkg/remoting/getty/listener.go

@@ -72,32 +72,30 @@ func (client *gettyClientHandler) OnOpen(session getty.Session) error {
 			sessionManager.ReleaseGettySession(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 (client *gettyClientHandler) OnClose(session getty.Session) {
+	log.Infof("OnClose session{%s} is closing......", session.Stat())
 	sessionManager.ReleaseGettySession(session)
 }
 
 func (client *gettyClientHandler) OnMessage(session getty.Session, pkg interface{}) {
 	ctx := context.Background()
 	log.Debugf("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()]
 		if processor != nil {
@@ -111,7 +109,7 @@ func (client *gettyClientHandler) OnMessage(session getty.Session, pkg interface
 }
 
 func (client *gettyClientHandler) OnCron(session getty.Session) {
-	GetGettyRemotingClient().SendAsyncRequest(message.HeartBeatMessagePing)
+	//GetGettyRemotingClient().SendAsyncRequest(message.HeartBeatMessagePing)
 }
 
 func (client *gettyClientHandler) RegisterProcessor(msgType message.MessageType, processor processor.RemotingProcessor) {

pkg/remoting/getty/rpc_client.go

@@ -58,7 +58,7 @@ func (c *RpcClient) init() {
 	for _, address := range addressList {
 		gettyClient := getty.NewTCPClient(
 			getty.WithServerAddress(address),
-			getty.WithConnectionNumber((int)(c.conf.GettyConfig.ConnectionNum)),
+			getty.WithConnectionNumber(c.conf.GettyConfig.ConnectionNum),
 			getty.WithReconnectInterval(c.conf.GettyConfig.ReconnectInterval),
 			getty.WithClientTaskPool(gxsync.NewTaskPoolSimple(0)),
 		)

pkg/remoting/getty/session_manager.go

@@ -27,17 +27,12 @@ import (
 
 var (
 	MAX_CHECK_ALIVE_RETRY = 600
-
-	CHECK_ALIVE_INTERNAL = 100
-
-	allSessions = sync.Map{}
-
+	CHECK_ALIVE_INTERNAL  = 100
+	allSessions           = sync.Map{}
 	// serverAddress -> rpc_client.Session -> bool
-	serverSessions = sync.Map{}
-
-	sessionSize int32 = 0
-
-	sessionManager = &GettySessionManager{}
+	serverSessions       = sync.Map{}
+	sessionSize    int32 = 0
+	sessionManager       = &GettySessionManager{}
 )
 
 type GettySessionManager struct{}

pkg/remoting/processor/client/client_heart_beat_processor_test.go

@@ -28,9 +28,9 @@ import (
 func TestClientHeartBeatProcessor(t *testing.T) {
 	// testcases
 	var tests = []struct {
-		name string // testcase name
-		rpcMsg message.RpcMessage // rpcMessage case
-		wantErr bool //want testcase err or not
+		name    string             // testcase name
+		rpcMsg  message.RpcMessage // rpcMessage case
+		wantErr bool               //want testcase err or not
 	}{
 		{
 			name: "chb-testcase1",

pkg/remoting/processor/client/client_on_response_processor_test.go

@@ -28,9 +28,9 @@ import (
 func TestClientOnResponseProcessor(t *testing.T) {
 	// testcases
 	var tests = []struct {
-		name string // testcase name
-		rpcMsg message.RpcMessage // rpcMessage case
-		wantErr bool //want testcase err or not
+		name    string             // testcase name
+		rpcMsg  message.RpcMessage // rpcMessage case
+		wantErr bool               //want testcase err or not
 	}{
 		{
 			name: "cor-testcase1-mergeResult",

pkg/remoting/processor/client/rm_branch_commit_processor_test.go

@@ -32,9 +32,9 @@ func TestRmBranchCommitProcessor(t *testing.T) {
 
 	// testcases
 	var tests = []struct {
-		name string // testcase name
-		rpcMsg message.RpcMessage // rpcMessage case
-		wantErr bool //want testcase err or not
+		name    string             // testcase name
+		rpcMsg  message.RpcMessage // rpcMessage case
+		wantErr bool               //want testcase err or not
 	}{
 		{
 			name: "rbc-testcase1-failure",