This is an automated email from the ASF dual-hosted git repository.
xingyue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 9a11740a refactor: abstract multi-database XA resource layer with
registry pattern (#1090)
9a11740a is described below
commit 9a11740a49ea547e3282bbd4375f1efc1cfe1734
Author: ThunGuo <[email protected]>
AuthorDate: Sat Apr 11 18:43:04 2026 +0800
refactor: abstract multi-database XA resource layer with registry pattern
(#1090)
* refactor: abstract multi-database XA resource layer with registry pattern
* Update pkg/datasource/sql/types/types_test.go
Co-authored-by: Copilot <[email protected]>
---------
Co-authored-by: Copilot <[email protected]>
---
pkg/datasource/sql/conn_xa.go | 23 +---
pkg/datasource/sql/conn_xa_test.go | 6 +-
pkg/datasource/sql/db.go | 7 +-
pkg/datasource/sql/types/types.go | 8 ++
pkg/datasource/sql/types/types_test.go | 10 +-
pkg/datasource/sql/xa/mysql_xa_connection.go | 40 ++++++-
pkg/datasource/sql/xa/oracle_xa_connection.go | 122 ++++++++++-----------
pkg/datasource/sql/xa/oracle_xa_connection_test.go | 122 ---------------------
pkg/datasource/sql/xa/postgres_xa_connection.go | 106 ++++++++++++++++++
pkg/datasource/sql/xa/xa_resource.go | 66 +++++++++++
pkg/datasource/sql/xa/xa_resource_factory.go | 48 --------
11 files changed, 298 insertions(+), 260 deletions(-)
diff --git a/pkg/datasource/sql/conn_xa.go b/pkg/datasource/sql/conn_xa.go
index f90216e2..5c7d1ba5 100644
--- a/pkg/datasource/sql/conn_xa.go
+++ b/pkg/datasource/sql/conn_xa.go
@@ -23,11 +23,8 @@ import (
"database/sql/driver"
"errors"
"fmt"
- "strings"
"time"
- "github.com/go-sql-driver/mysql"
-
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/xa"
"seata.apache.org/seata-go/v2/pkg/tm"
@@ -43,6 +40,7 @@ type XAConn struct {
tx driver.Tx
xaResource xa.XAResource
+ xaErrorClassifier xa.XAErrorClassifier
xaBranchXid *XABranchXid
xaActive bool
rollBacked bool
@@ -251,6 +249,7 @@ func (c *XAConn) start(ctx context.Context) error {
return fmt.Errorf("create xa xid:%s resoruce err:%w",
c.txCtx.XID, err)
}
c.xaResource = xaResource
+ c.xaErrorClassifier = xa.CreateErrorClassifier(c.dbType)
if err := c.xaResource.Start(ctx, c.xaBranchXid.String(),
xa.TMNoFlags); err != nil {
return fmt.Errorf("xa xid %s resource connection start err:%w",
c.txCtx.XID, err)
@@ -309,7 +308,7 @@ func (c *XAConn) Rollback(ctx context.Context) error {
if err := c.xaResource.End(ctx, c.xaBranchXid.String(),
xa.TMFail); err != nil {
// Handle XAER_RMFAIL exception - check if it's already
ended
//expected error: Error 1399 (XAE07): XAER_RMFAIL: The
command cannot be executed when global transaction is in the IDLE state
- if isXAER_RMFAILAlreadyEnded(err) {
+ if c.xaErrorClassifier.IsAlreadyEnded(err) {
// If already ended, continue with rollback
log.Infof("XA branch already ended, continuing
with rollback for xid: %s", c.txCtx.XID)
} else {
@@ -418,19 +417,3 @@ func (c *XAConn) XaRollback(ctx context.Context, xaXid
XAXid) error {
c.releaseIfNecessary()
return err
}
-
-// isXAER_RMFAILAlreadyEnded checks if the XAER_RMFAIL error indicates the XA
branch is already ended
-// expected error: Error 1399 (XAE07): XAER_RMFAIL: The command cannot be
executed when global transaction is in the IDLE state
-func isXAER_RMFAILAlreadyEnded(err error) bool {
- if err == nil {
- return false
- }
- if mysqlErr, ok := err.(*mysql.MySQLError); ok {
- if mysqlErr.Number == types.ErrCodeXAER_RMFAIL_IDLE {
- return strings.Contains(mysqlErr.Message, "IDLE state")
|| strings.Contains(mysqlErr.Message, "already ended")
- }
- }
- // TODO: handle other DB errors
-
- return false
-}
diff --git a/pkg/datasource/sql/conn_xa_test.go
b/pkg/datasource/sql/conn_xa_test.go
index fb40084a..6d5e26fc 100644
--- a/pkg/datasource/sql/conn_xa_test.go
+++ b/pkg/datasource/sql/conn_xa_test.go
@@ -36,6 +36,7 @@ import (
"seata.apache.org/seata-go/v2/pkg/datasource/sql/exec"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/mock"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/xa"
"seata.apache.org/seata-go/v2/pkg/protocol/branch"
"seata.apache.org/seata-go/v2/pkg/tm"
)
@@ -378,8 +379,9 @@ func TestXAConn_Rollback_XAER_RMFAIL(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- if got := isXAER_RMFAILAlreadyEnded(tt.err); got !=
tt.want {
- t.Errorf("isXAER_RMFAILAlreadyEnded() = %v,
want %v", got, tt.want)
+ classifier := &xa.MysqlXAErrorClassifier{}
+ if got := classifier.IsAlreadyEnded(tt.err); got !=
tt.want {
+
t.Errorf("MysqlXAErrorClassifier.IsAlreadyEnded() = %v, want %v", got, tt.want)
}
})
}
diff --git a/pkg/datasource/sql/db.go b/pkg/datasource/sql/db.go
index 39a85fe2..cc019283 100644
--- a/pkg/datasource/sql/db.go
+++ b/pkg/datasource/sql/db.go
@@ -221,7 +221,7 @@ func (db *DBResource) ConnectionForXA(ctx context.Context,
xaXid XAXid) (*XAConn
if err != nil {
return nil, fmt.Errorf("get xa new connection failure, xid:%s,
err:%v", xaXid.String(), err)
}
- xaResource, err := xa.CreateXAResource(newDriverConn, types.DBTypeMySQL)
+ xaResource, err := xa.CreateXAResource(newDriverConn, db.dbType)
if err != nil {
return nil, fmt.Errorf("create xa resoruce err:%w", err)
}
@@ -230,8 +230,9 @@ func (db *DBResource) ConnectionForXA(ctx context.Context,
xaXid XAXid) (*XAConn
targetConn: newDriverConn,
res: db,
},
- xaBranchXid: XaIdBuild(xaXid.GetGlobalXid(),
xaXid.GetBranchId()),
- xaResource: xaResource,
+ xaBranchXid: XaIdBuild(xaXid.GetGlobalXid(),
xaXid.GetBranchId()),
+ xaResource: xaResource,
+ xaErrorClassifier: xa.CreateErrorClassifier(db.dbType),
}
return xaConn, nil
}
diff --git a/pkg/datasource/sql/types/types.go
b/pkg/datasource/sql/types/types.go
index d85cea32..74598ce1 100644
--- a/pkg/datasource/sql/types/types.go
+++ b/pkg/datasource/sql/types/types.go
@@ -96,6 +96,14 @@ func ParseDBType(driverName string) DBType {
switch strings.ToLower(driverName) {
case "mysql":
return DBTypeMySQL
+ case "postgres", "postgresql", "pgx":
+ return DBTypePostgreSQL
+ case "oracle", "godror", "go-ora":
+ return DBTypeOracle
+ case "sqlserver", "mssql":
+ return DBTypeSQLServer
+ case "mariadb":
+ return DBTypeMARIADB
default:
return DBTypeUnknown
}
diff --git a/pkg/datasource/sql/types/types_test.go
b/pkg/datasource/sql/types/types_test.go
index cd48c16f..b15dbd8e 100644
--- a/pkg/datasource/sql/types/types_test.go
+++ b/pkg/datasource/sql/types/types_test.go
@@ -114,7 +114,15 @@ func TestParseDBType(t *testing.T) {
{"mysql", "mysql", DBTypeMySQL},
{"MySQL uppercase", "MySQL", DBTypeMySQL},
{"MYSQL", "MYSQL", DBTypeMySQL},
- {"postgres", "postgres", DBTypeUnknown},
+ {"postgres", "postgres", DBTypePostgreSQL},
+ {"postgresql", "postgresql", DBTypePostgreSQL},
+ {"pgx", "pgx", DBTypePostgreSQL},
+ {"oracle", "oracle", DBTypeOracle},
+ {"godror", "godror", DBTypeOracle},
+ {"go-ora", "go-ora", DBTypeOracle},
+ {"sqlserver", "sqlserver", DBTypeSQLServer},
+ {"mssql", "mssql", DBTypeSQLServer},
+ {"mariadb", "mariadb", DBTypeMARIADB},
{"unknown", "unknown", DBTypeUnknown},
{"empty", "", DBTypeUnknown},
}
diff --git a/pkg/datasource/sql/xa/mysql_xa_connection.go
b/pkg/datasource/sql/xa/mysql_xa_connection.go
index f60ba492..601a27dd 100644
--- a/pkg/datasource/sql/xa/mysql_xa_connection.go
+++ b/pkg/datasource/sql/xa/mysql_xa_connection.go
@@ -26,17 +26,51 @@ import (
"strings"
"time"
+ "github.com/go-sql-driver/mysql"
+
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
"seata.apache.org/seata-go/v2/pkg/util/log"
)
-type MysqlXAConn struct {
- driver.Conn
+func init() {
+ RegisterXAResourceFactory(types.DBTypeMySQL, &mysqlXAResourceFactory{})
}
-func NewMysqlXaConn(conn driver.Conn) *MysqlXAConn {
+// mysqlXAResourceFactory creates MySQL-specific XA resources and error
classifiers.
+type mysqlXAResourceFactory struct{}
+
+func (f *mysqlXAResourceFactory) CreateXAResource(conn driver.Conn) XAResource
{
return &MysqlXAConn{Conn: conn}
}
+func (f *mysqlXAResourceFactory) CreateErrorClassifier() XAErrorClassifier {
+ return &MysqlXAErrorClassifier{}
+}
+
+// MysqlXAErrorClassifier classifies MySQL-specific XA errors.
+type MysqlXAErrorClassifier struct{}
+
+// IsAlreadyEnded checks if the XAER_RMFAIL error indicates the XA branch is
already ended.
+// Expected error: Error 1399 (XAE07): XAER_RMFAIL: The command cannot be
executed
+// when global transaction is in the IDLE state
+func (c *MysqlXAErrorClassifier) IsAlreadyEnded(err error) bool {
+ if err == nil {
+ return false
+ }
+ var mysqlErr *mysql.MySQLError
+ if errors.As(err, &mysqlErr) {
+ if mysqlErr.Number == types.ErrCodeXAER_RMFAIL_IDLE {
+ return strings.Contains(mysqlErr.Message, "IDLE state")
|| strings.Contains(mysqlErr.Message, "already ended")
+ }
+ }
+ return false
+}
+
+// MysqlXAConn implements XAResource for MySQL using native XA SQL statements.
+type MysqlXAConn struct {
+ driver.Conn
+}
+
func (c *MysqlXAConn) Commit(ctx context.Context, xid string, onePhase bool)
error {
log.Infof("xa branch commit, xid %s", xid)
diff --git a/pkg/datasource/sql/xa/oracle_xa_connection.go
b/pkg/datasource/sql/xa/oracle_xa_connection.go
index df03920f..8ba132a0 100644
--- a/pkg/datasource/sql/xa/oracle_xa_connection.go
+++ b/pkg/datasource/sql/xa/oracle_xa_connection.go
@@ -20,92 +20,92 @@ package xa
import (
"context"
"database/sql/driver"
- "strings"
+ "fmt"
"time"
- _ "github.com/sijms/go-ora/v2"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
)
-type OracleXAConn struct {
- driver.Conn
+func init() {
+ RegisterXAResourceFactory(types.DBTypeOracle,
&oracleXAResourceFactory{})
}
-func (c *OracleXAConn) Commit(xid string, onePhase bool) error {
- var sb strings.Builder
- sb.WriteString("XA COMMIT ")
- sb.WriteString(xid)
- if onePhase {
- sb.WriteString(" ONE PHASE")
- }
+type oracleXAResourceFactory struct{}
- conn, _ := c.Conn.(driver.ExecerContext)
- _, err := conn.ExecContext(context.TODO(), sb.String(), nil)
- return err
+func (f *oracleXAResourceFactory) CreateXAResource(conn driver.Conn)
XAResource {
+ return &OracleXAConn{Conn: conn}
}
-func (c *OracleXAConn) End(xid string, flags int) error {
- var sb strings.Builder
- sb.WriteString("XA END ")
- sb.WriteString(xid)
-
- conn, _ := c.Conn.(driver.ExecerContext)
- _, err := conn.ExecContext(context.TODO(), sb.String(), nil)
- return err
+func (f *oracleXAResourceFactory) CreateErrorClassifier() XAErrorClassifier {
+ return &OracleXAErrorClassifier{}
}
-func (c *OracleXAConn) Forget(xid string) error {
- // TODO implement me
- panic("implement me")
+// OracleXAErrorClassifier classifies Oracle-specific XA errors.
+type OracleXAErrorClassifier struct{}
+
+func (c *OracleXAErrorClassifier) IsAlreadyEnded(err error) bool {
+ // TODO: check ORA-24756 (transaction does not exist) / ORA-24761
(rolled back)
+ return false
}
-func (c *OracleXAConn) GetTransactionTimeout() time.Duration {
- // TODO implement me
- panic("implement me")
+// OracleXAConn implements XAResource for Oracle using the DBMS_XA PL/SQL
package.
+//
+// Oracle does NOT support MySQL-style XA SQL statements.
+// All operations go through PL/SQL anonymous blocks calling DBMS_XA:
+// - DBMS_XA.XA_START / XA_END / XA_PREPARE / XA_COMMIT / XA_ROLLBACK
+// - XID type: DBMS_XA_XID(formatid NUMBER, gtrid RAW(64), bqual RAW(64))
+// - Recovery: DBA_PENDING_TRANSACTIONS or DBMS_XA.XA_RECOVER
+// - Requires: GRANT EXECUTE ON DBMS_XA TO <user>
+type OracleXAConn struct {
+ driver.Conn
}
-func (c *OracleXAConn) IsSameRM(resource XAResource) bool {
- // TODO implement me
- panic("implement me")
+func (c *OracleXAConn) Start(ctx context.Context, xid string, flags int) error
{
+ log.Infof("xa branch start (oracle), xid %s", xid)
+ // TODO: DBMS_XA.XA_START via PL/SQL block, map xid to DBMS_XA_XID
+ return fmt.Errorf("Oracle XA start not yet implemented")
}
-func (c *OracleXAConn) XAPrepare(xid string) (int, error) {
- var sb strings.Builder
- sb.WriteString("XA PREPARE ")
- sb.WriteString(xid)
+func (c *OracleXAConn) End(ctx context.Context, xid string, flags int) error {
+ log.Infof("xa branch end (oracle), xid %s", xid)
+ // TODO: DBMS_XA.XA_END via PL/SQL block
+ return fmt.Errorf("Oracle XA end not yet implemented")
+}
- conn, _ := c.Conn.(driver.ExecerContext)
- if _, err := conn.ExecContext(context.TODO(), sb.String(), nil); err !=
nil {
- return -1, err
- }
- return 0, nil
+func (c *OracleXAConn) XAPrepare(ctx context.Context, xid string) error {
+ log.Infof("xa branch prepare (oracle), xid %s", xid)
+ // TODO: DBMS_XA.XA_PREPARE via PL/SQL block
+ return fmt.Errorf("Oracle XA prepare not yet implemented")
}
-func (c *OracleXAConn) Recover(flag int) []string {
- // TODO implement me
- panic("implement me")
+func (c *OracleXAConn) Commit(ctx context.Context, xid string, onePhase bool)
error {
+ log.Infof("xa branch commit (oracle), xid %s, onePhase %v", xid,
onePhase)
+ // TODO: DBMS_XA.XA_COMMIT via PL/SQL block
+ return fmt.Errorf("Oracle XA commit not yet implemented")
}
-func (c *OracleXAConn) Rollback(xid string) error {
- var sb strings.Builder
- sb.WriteString("XA ROLLBACK ")
- sb.WriteString(xid)
+func (c *OracleXAConn) Rollback(ctx context.Context, xid string) error {
+ log.Infof("xa branch rollback (oracle), xid %s", xid)
+ // TODO: DBMS_XA.XA_ROLLBACK via PL/SQL block
+ return fmt.Errorf("Oracle XA rollback not yet implemented")
+}
- conn, _ := c.Conn.(driver.ExecerContext)
- _, err := conn.ExecContext(context.TODO(), sb.String(), nil)
- return err
+func (c *OracleXAConn) Recover(ctx context.Context, flag int) ([]string,
error) {
+ if (flag & TMStartRScan) == 0 {
+ return nil, nil
+ }
+ // TODO: SELECT globalid, branchid FROM DBA_PENDING_TRANSACTIONS
+ return nil, fmt.Errorf("Oracle XA recover not yet implemented")
}
-func (c *OracleXAConn) SetTransactionTimeout(duration time.Duration) bool {
- // TODO implement me
- panic("implement me")
+func (c *OracleXAConn) Forget(ctx context.Context, xid string) error {
+ // TODO: DBMS_XA.XA_FORGET via PL/SQL block
+ return fmt.Errorf("Oracle XA forget not yet implemented")
}
-func (c *OracleXAConn) Start(xid string, flags int) error {
- var sb strings.Builder
- sb.WriteString("XA START")
- sb.WriteString(xid)
+func (c *OracleXAConn) GetTransactionTimeout() time.Duration { return 0 }
- conn, _ := c.Conn.(driver.ExecerContext)
- _, err := conn.ExecContext(context.TODO(), sb.String(), nil)
- return err
-}
+func (c *OracleXAConn) IsSameRM(ctx context.Context, resource XAResource) bool
{ return false }
+
+func (c *OracleXAConn) SetTransactionTimeout(duration time.Duration) bool {
return false }
diff --git a/pkg/datasource/sql/xa/oracle_xa_connection_test.go
b/pkg/datasource/sql/xa/oracle_xa_connection_test.go
deleted file mode 100644
index 6b8471e7..00000000
--- a/pkg/datasource/sql/xa/oracle_xa_connection_test.go
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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 xa
-
-import (
- "database/sql/driver"
- "testing"
- "time"
-)
-
-func TestOracleXAConn_Commit(t *testing.T) {
- type fields struct {
- Conn driver.Conn
- }
- type args struct {
- xid string
- onePhase bool
- }
- var tests []struct {
- name string
- fields fields
- args args
- wantErr bool
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- c := &OracleXAConn{
- Conn: tt.fields.Conn,
- }
- if err := c.Commit(tt.args.xid, tt.args.onePhase); (err
!= nil) != tt.wantErr {
- t.Errorf("Commit() error = %v, wantErr %v",
err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOracleXAConn_End(t *testing.T) {
- type fields struct {
- Conn driver.Conn
- }
- type args struct {
- xid string
- flags int
- }
- var tests []struct {
- name string
- fields fields
- args args
- wantErr bool
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- c := &OracleXAConn{
- Conn: tt.fields.Conn,
- }
- if err := c.End(tt.args.xid, tt.args.flags); (err !=
nil) != tt.wantErr {
- t.Errorf("End() error = %v, wantErr %v", err,
tt.wantErr)
- }
- })
- }
-}
-
-func TestOracleXAConn_Forget(t *testing.T) {
- type fields struct {
- Conn driver.Conn
- }
- type args struct {
- xid string
- }
- var tests []struct {
- name string
- fields fields
- args args
- wantErr bool
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- c := &OracleXAConn{
- Conn: tt.fields.Conn,
- }
- if err := c.Forget(tt.args.xid); (err != nil) !=
tt.wantErr {
- t.Errorf("Forget() error = %v, wantErr %v",
err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOracleXAConn_GetTransactionTimeout(t *testing.T) {
- type fields struct {
- Conn driver.Conn
- }
- var tests []struct {
- name string
- fields fields
- want time.Duration
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- c := &OracleXAConn{
- Conn: tt.fields.Conn,
- }
- if got := c.GetTransactionTimeout(); got != tt.want {
- t.Errorf("GetTransactionTimeout() = %v, want
%v", got, tt.want)
- }
- })
- }
-}
diff --git a/pkg/datasource/sql/xa/postgres_xa_connection.go
b/pkg/datasource/sql/xa/postgres_xa_connection.go
new file mode 100644
index 00000000..a1a0fb51
--- /dev/null
+++ b/pkg/datasource/sql/xa/postgres_xa_connection.go
@@ -0,0 +1,106 @@
+/*
+ * 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 xa
+
+import (
+ "context"
+ "database/sql/driver"
+ "fmt"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+func init() {
+ RegisterXAResourceFactory(types.DBTypePostgreSQL,
&postgresXAResourceFactory{})
+}
+
+type postgresXAResourceFactory struct{}
+
+func (f *postgresXAResourceFactory) CreateXAResource(conn driver.Conn)
XAResource {
+ return &PostgresXAConn{Conn: conn}
+}
+
+func (f *postgresXAResourceFactory) CreateErrorClassifier() XAErrorClassifier {
+ return &PostgresXAErrorClassifier{}
+}
+
+// PostgresXAErrorClassifier classifies PostgreSQL-specific XA errors.
+type PostgresXAErrorClassifier struct{}
+
+func (c *PostgresXAErrorClassifier) IsAlreadyEnded(err error) bool {
+ // TODO: check pgconn.PgError SQLSTATE "42704" / "55000"
+ return false
+}
+
+// PostgresXAConn implements XAResource for PostgreSQL using native 2PC.
+//
+// Key differences from MySQL XA:
+// - Start/End are no-ops (PostgreSQL uses regular BEGIN, no XA START/END).
+// - XAPrepare → PREPARE TRANSACTION 'xid'
+// - Commit → COMMIT PREPARED 'xid'
+// - Rollback → ROLLBACK PREPARED 'xid'
+// - Recover → SELECT gid FROM pg_prepared_xacts
+// - Requires max_prepared_transactions > 0 in postgresql.conf.
+type PostgresXAConn struct {
+ driver.Conn
+}
+
+func (c *PostgresXAConn) Start(ctx context.Context, xid string, flags int)
error {
+ log.Infof("xa branch start (postgres no-op), xid %s", xid)
+ return nil
+}
+
+func (c *PostgresXAConn) End(ctx context.Context, xid string, flags int) error
{
+ log.Infof("xa branch end (postgres no-op), xid %s", xid)
+ return nil
+}
+
+func (c *PostgresXAConn) XAPrepare(ctx context.Context, xid string) error {
+ // TODO: PREPARE TRANSACTION 'xid'
+ return fmt.Errorf("PostgreSQL XA prepare not yet implemented")
+}
+
+func (c *PostgresXAConn) Commit(ctx context.Context, xid string, onePhase
bool) error {
+ // TODO: COMMIT PREPARED 'xid' (onePhase=true → regular commit by upper
layer)
+ return fmt.Errorf("PostgreSQL XA commit not yet implemented")
+}
+
+func (c *PostgresXAConn) Rollback(ctx context.Context, xid string) error {
+ // TODO: ROLLBACK PREPARED 'xid'
+ return fmt.Errorf("PostgreSQL XA rollback not yet implemented")
+}
+
+func (c *PostgresXAConn) Recover(ctx context.Context, flag int) ([]string,
error) {
+ if (flag & TMStartRScan) == 0 {
+ return nil, nil
+ }
+ // TODO: SELECT gid FROM pg_prepared_xacts WHERE database =
current_database()
+ return nil, fmt.Errorf("PostgreSQL XA recover not yet implemented")
+}
+
+func (c *PostgresXAConn) Forget(ctx context.Context, xid string) error {
+ return fmt.Errorf("PostgreSQL does not support XA forget")
+}
+
+func (c *PostgresXAConn) GetTransactionTimeout() time.Duration { return 0 }
+
+func (c *PostgresXAConn) IsSameRM(ctx context.Context, resource XAResource)
bool { return false }
+
+func (c *PostgresXAConn) SetTransactionTimeout(duration time.Duration) bool {
return false }
diff --git a/pkg/datasource/sql/xa/xa_resource.go
b/pkg/datasource/sql/xa/xa_resource.go
index db26a643..c37fc118 100644
--- a/pkg/datasource/sql/xa/xa_resource.go
+++ b/pkg/datasource/sql/xa/xa_resource.go
@@ -19,7 +19,11 @@ package xa
import (
"context"
+ "database/sql/driver"
+ "fmt"
"time"
+
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
)
const (
@@ -57,6 +61,8 @@ const (
XAOk = 0
)
+// XAResource defines the contract for XA transaction operations across
different databases.
+// Each database (MySQL, PostgreSQL, Oracle, etc.) provides its own
implementation.
type XAResource interface {
Commit(ctx context.Context, xid string, onePhase bool) error
End(ctx context.Context, xid string, flags int) error
@@ -69,3 +75,63 @@ type XAResource interface {
SetTransactionTimeout(duration time.Duration) bool
Start(ctx context.Context, xid string, flags int) error
}
+
+// XAErrorClassifier abstracts database-specific XA error classification.
+// This allows the upper layer (conn_xa.go) to handle XA errors without
+// importing database-specific driver packages.
+type XAErrorClassifier interface {
+ // IsAlreadyEnded checks if the error indicates the XA branch is
already ended.
+ // For MySQL: XAER_RMFAIL with IDLE state (error 1399).
+ // For PostgreSQL: transaction already committed/rolled back.
+ // For Oracle: ORA-24756 (transaction does not exist).
+ IsAlreadyEnded(err error) bool
+}
+
+// defaultErrorClassifier is a no-op classifier that never matches any error.
+// Used as a fallback when no database-specific classifier is registered.
+type defaultErrorClassifier struct{}
+
+func (c *defaultErrorClassifier) IsAlreadyEnded(err error) bool { return false
}
+
+// XAResourceFactory creates database-specific XA resources and error
classifiers.
+type XAResourceFactory interface {
+ // CreateXAResource creates a new XAResource for the given driver
connection.
+ CreateXAResource(conn driver.Conn) XAResource
+ // CreateErrorClassifier creates a database-specific error classifier.
+ CreateErrorClassifier() XAErrorClassifier
+}
+
+// registry holds registered XAResourceFactory instances per DBType.
+var registry = map[types.DBType]XAResourceFactory{}
+
+// RegisterXAResourceFactory registers a factory for the given database type.
+// Each database driver package should call this in its init() function.
+func RegisterXAResourceFactory(dbType types.DBType, factory XAResourceFactory)
{
+ registry[dbType] = factory
+}
+
+// GetXAResourceFactory returns the registered factory for the given database
type.
+func GetXAResourceFactory(dbType types.DBType) (XAResourceFactory, bool) {
+ f, ok := registry[dbType]
+ return f, ok
+}
+
+// CreateXAResource creates an XAResource for the given database type and
connection.
+// It uses the registered factory for the database type.
+func CreateXAResource(conn driver.Conn, dbType types.DBType) (XAResource,
error) {
+ factory, ok := GetXAResourceFactory(dbType)
+ if !ok {
+ return nil, fmt.Errorf("no XA resource factory registered for
db type: %s", dbType.String())
+ }
+ return factory.CreateXAResource(conn), nil
+}
+
+// CreateErrorClassifier creates an XAErrorClassifier for the given database
type.
+// Returns a default no-op classifier if no factory is registered.
+func CreateErrorClassifier(dbType types.DBType) XAErrorClassifier {
+ factory, ok := GetXAResourceFactory(dbType)
+ if !ok {
+ return &defaultErrorClassifier{}
+ }
+ return factory.CreateErrorClassifier()
+}
diff --git a/pkg/datasource/sql/xa/xa_resource_factory.go
b/pkg/datasource/sql/xa/xa_resource_factory.go
deleted file mode 100644
index ac6c1a6e..00000000
--- a/pkg/datasource/sql/xa/xa_resource_factory.go
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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 xa
-
-import (
- "database/sql/driver"
- "fmt"
-
- "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
- "seata.apache.org/seata-go/v2/pkg/util/log"
-)
-
-// CreateXAResource create a connection for xa with the different db type.
-// Such as mysql, oracle, MARIADB, POSTGRESQL
-func CreateXAResource(conn driver.Conn, dbType types.DBType) (XAResource,
error) {
- var err error
- var xaConnection XAResource
- switch dbType {
- case types.DBTypeMySQL:
- xaConnection = NewMysqlXaConn(conn)
- case types.DBTypeOracle:
- case types.DBTypePostgreSQL:
- default:
- err = fmt.Errorf("not support db type for :%s", dbType.String())
- }
-
- if err != nil {
- log.Errorf(err.Error())
- return nil, err
- }
-
- return xaConnection, nil
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
