feat: support xa protocol (#9)

Author: dk-lockdown

connection.go

@@ -22,11 +22,11 @@ import (
 
 	"github.com/cectc/dbpack/pkg/log"
 	"github.com/cectc/hptx/pkg/config"
+	"github.com/cectc/hptx/pkg/constant"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/parser/ast"
 )
 
-const XID = "XID"
 const GlobalLock = "GlobalLock"
 
 type connCtx struct {
@@ -588,7 +588,7 @@ func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver
 		}
 	}
 
-	val := ctx.Value(XID)
+	val := ctx.Value(constant.XID)
 	if val != nil {
 		if xid, ok := val.(string); ok {
 			mc.ctx = &connCtx{

datasource_manager.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cectc/dbpack/pkg/dt/api"
 	"github.com/cectc/dbpack/pkg/log"
 	"github.com/cectc/hptx/pkg/core"
+	"github.com/cectc/hptx/pkg/resource"
 	"github.com/pkg/errors"
 )
 
@@ -25,11 +26,7 @@ func init() {
 	}
 }
 
-func GetDataSourceManager() DataSourceManager {
-	return dataSourceManager
-}
-
-func RegisterResource(dsn string) {
+func RegisterATResource(dsn string) {
 	cfg, err := ParseDSN(dsn)
 	if err == nil {
 		c := &connector{
@@ -38,6 +35,7 @@ func RegisterResource(dsn string) {
 		dataSourceManager.ResourceCache[c.cfg.DBName] = c
 		InitTableMetaCache(c.cfg.DBName)
 	}
+	resource.InitATBranchResource(dataSourceManager)
 }
 
 func (resourceManager DataSourceManager) GetConnection(resourceID string) *mysqlConn {

go.mod

@@ -4,7 +4,7 @@ go 1.13
 
 require (
 	github.com/cectc/dbpack v0.3.1
-	github.com/cectc/hptx v1.0.4
+	github.com/cectc/hptx v1.0.5
 	github.com/google/go-cmp v0.5.7
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pingcap/parser v0.0.0-20210831085004-b5390aa83f65

go.sum

@@ -121,8 +121,8 @@ github.com/carlmjohnson/flagext v0.21.0/go.mod h1:Eenv0epIUAr4NuedNmkzI8WmBmjIxZ
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
 github.com/cectc/dbpack v0.3.1 h1:bkIS0dMecGT14EeD7/gBgroyy/zfO27Dk4iEbfDUTkI=
 github.com/cectc/dbpack v0.3.1/go.mod h1:xofYTSvQFzwaN5L3KrEh8/chYg4pcbCo/+UVg1y99Hs=
-github.com/cectc/hptx v1.0.4 h1:x/bssMsYmIDMKI5kkeeOIJftPuRfNENZp5O95WOvgqw=
-github.com/cectc/hptx v1.0.4/go.mod h1:6UF0JJtfOksYnUL9gYCUgufb95PT0e2KD/WYN4LPgzk=
+github.com/cectc/hptx v1.0.5 h1:UrT8eqSTU4vEDpuNFiiOBmaZCe/UE3s+3YHy/oGbob8=
+github.com/cectc/hptx v1.0.5/go.mod h1:6UF0JJtfOksYnUL9gYCUgufb95PT0e2KD/WYN4LPgzk=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cenkalti/backoff/v4 v4.0.2/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=

xa_resource_manager.go

@@ -0,0 +1,50 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/cectc/dbpack/pkg/dt/api"
+	"github.com/cectc/hptx/pkg/resource"
+	"github.com/pkg/errors"
+)
+
+var xaResourceManager ResourceManager
+
+type ResourceManager struct {
+	db *sql.DB
+}
+
+func RegisterXAResource(db *sql.DB) {
+	xaResourceManager = ResourceManager{db: db}
+	resource.InitXABranchResource(xaResourceManager)
+}
+
+func (resourceManager ResourceManager) Commit(ctx context.Context, bs *api.BranchSession) (api.BranchSession_BranchStatus, error) {
+	_, err := resourceManager.db.ExecContext(ctx, fmt.Sprintf("XA COMMIT '%s'", bs.BranchID))
+	if err != nil {
+		var mysqlErr *MySQLError
+		if errors.As(err, &mysqlErr) {
+			if mysqlErr.Number == 1399 {
+				return api.Complete, nil
+			}
+		}
+		return bs.Status, err
+	}
+	return api.Complete, nil
+}
+
+func (resourceManager ResourceManager) Rollback(ctx context.Context, bs *api.BranchSession) (api.BranchSession_BranchStatus, error) {
+	_, err := resourceManager.db.ExecContext(ctx, fmt.Sprintf("XA ROLLBACK '%s'", bs.BranchID))
+	if err != nil {
+		var mysqlErr *MySQLError
+		if errors.As(err, &mysqlErr) {
+			if mysqlErr.Number == 1399 {
+				return api.Complete, nil
+			}
+		}
+		return bs.Status, err
+	}
+	return api.Complete, nil
+}