Merge pull request #9 from catcherwong/barrier

Support Barrier

Author: yedf2

src/Dtmcli.Tests/BranchBarrierTests.cs

@@ -0,0 +1,83 @@
+using Apps72.Dev.Data.DbMocker;
+using Apps72.Dev.Data.DbMocker.Data;
+using Xunit;
+using Dapper;
+using System;
+using System.Threading.Tasks;
+using Moq;
+using System.Linq;
+
+namespace Dtmcli.Tests
+{
+    public class BranchBarrierTests
+    {
+        [Theory]
+        [InlineData("cancel", "try")]
+        [InlineData("compensate", "action")]
+        public async void Call_Should_Not_Trigger_When_IsNullCompensation(string op, string origin)
+        {
+            var branchBarrier = new BranchBarrier("tcc", "gid", "bid", op);
+
+            var conn = GetDbConnection();
+
+            // mock originAffected = 1
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(origin)).ReturnsScalar(cmd => 1);
+
+            // mock currentAffected != 0
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(op)).ReturnsScalar(cmd => 1);
+
+            var mockBusiCall = new Mock<Func<System.Data.Common.DbTransaction, Task>>();
+
+            await branchBarrier.Call(conn, mockBusiCall.Object);
+
+            mockBusiCall.Verify(x => x.Invoke(It.IsAny<System.Data.Common.DbTransaction>()), Times.Never);
+        }
+
+        [Theory]
+        [InlineData("other1", "other2")]
+        public async void Call_Should_Not_Trigger_When_IsDuplicateOrPend(string op, string origin)
+        {
+            var branchBarrier = new BranchBarrier("tcc", "gid", "bid", op);
+
+            var conn = GetDbConnection();
+
+            // mock originAffected = 0
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(origin)).ReturnsScalar(cmd => 0);
+
+            // mock currentAffected = 0
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(op)).ReturnsScalar(cmd => 0);
+
+            var mockBusiCall = new Mock<Func<System.Data.Common.DbTransaction, Task>>();
+
+            await branchBarrier.Call(conn, mockBusiCall.Object);
+
+            mockBusiCall.Verify(x => x.Invoke(It.IsAny<System.Data.Common.DbTransaction>()), Times.Never);
+        }
+
+        [Theory]
+        [InlineData("cancel", "try")]
+        [InlineData("compensate", "action")]
+        public async void Call_Should_Trigger_When_IsNotNullCompensation_And_DuplicateOrPend(string op, string origin)
+        {
+            var branchBarrier = new BranchBarrier("tcc", "gid", "bid", op);
+
+            var conn = GetDbConnection();
+
+            // mock originAffected = 0
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(origin)).ReturnsScalar(cmd => 0);
+
+            // mock currentAffected > 0
+            conn.Mocks.When(cmd => cmd.Parameters.AsList().Select(x => x.Value).Contains(op)).ReturnsScalar(cmd => 1);
+
+            var mockBusiCall = new Mock<Func<System.Data.Common.DbTransaction, Task>>();
+
+            await branchBarrier.Call(conn, mockBusiCall.Object);
+
+            mockBusiCall.Verify(x => x.Invoke(It.IsAny<System.Data.Common.DbTransaction>()), Times.Once);
+        }
+
+        private MockDbConnection GetDbConnection() => new MockDbConnection();
+
+
+    }
+}

src/Dtmcli.Tests/BranchIDGenTests.cs

@@ -0,0 +1,48 @@
+using Dtmcli.DtmImp;
+using System;
+using Xunit;
+
+namespace Dtmcli.Tests
+{
+    public class BranchIDGenTests
+    {
+        [Fact]
+        public void TestNewSubBranchID()
+        {
+            var b = new BranchIDGen("");
+
+            // 01,...,09
+            for (int i = 0; i < 9; i++)
+            {
+                var n = b.NewSubBranchID();
+                Assert.Equal($"0{i+1}", n);
+            }
+
+            // 10~98
+            for (int i = 9; i < 99; i++)
+            {
+                var n = b.NewSubBranchID();
+                Assert.Equal($"{i + 1}", n);
+            }
+
+            // 99~
+            Assert.Throws<ArgumentException>(() => b.NewSubBranchID());
+        }
+
+        [Fact]
+        public void NewSubBranchID_With_BranchId_Should_Succeed()
+        {
+            var b = new BranchIDGen("ss");
+            var n = b.NewSubBranchID();
+            Assert.Equal($"ss01", n);
+        }
+
+        [Fact]
+        public void NewSubBranchID_With_BranchId_Should_Throw_Exception()
+        {
+            var b = new BranchIDGen("sssssssssssssssssssss");
+            Assert.Throws<ArgumentException>(() => b.NewSubBranchID());
+        }
+
+    }
+}

src/Dtmcli.Tests/DbSpecialTests.cs

@@ -0,0 +1,23 @@
+using Xunit;
+
+namespace Dtmcli.Tests
+{
+    public class DbSpecialTests
+    {
+        [Fact]
+        public void TestDbSpecial()
+        {
+            DtmImp.DbSpecialDelegate.Instance.SetCurrentDBType("mysql");
+
+            var mysql = DtmImp.DbSpecialDelegate.Instance.GetDBSpecial();
+            Assert.Equal("xa start 'xa1'", mysql.GetXaSQL("start", "xa1"));
+            Assert.Equal("insert ignore into a(f) values(@f)", mysql.GetInsertIgnoreTemplate("a(f) values(@f)", "c"));
+
+            DtmImp.DbSpecialDelegate.Instance.SetCurrentDBType("postgres");
+
+            var postgres = DtmImp.DbSpecialDelegate.Instance.GetDBSpecial();
+            Assert.Equal("begin", postgres.GetXaSQL("start", "xa1"));
+            Assert.Equal("insert into a(f) values(@f) on conflict ON CONSTRAINT c do nothing", postgres.GetInsertIgnoreTemplate("a(f) values(@f)", "c"));
+        }
+    }
+}

src/Dtmcli.Tests/Dtmcli.Tests.csproj

@@ -8,6 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="DbMocker" Version="1.21.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.11.0" />
     <PackageReference Include="Moq" Version="4.16.1" />
     <PackageReference Include="xunit" Version="2.4.1" />

src/Dtmcli/Barrier/BranchBarrier.cs

@@ -0,0 +1,138 @@
+using Dapper;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Data.Common;
+using System.Threading.Tasks;
+
+namespace Dtmcli
+{
+    public class BranchBarrier
+    {
+        public BranchBarrier(string transType, string gid, string branchID, string op, ILogger logger = null)
+        {
+            this.TransType = transType;
+            this.Gid = gid;
+            this.BranchID = branchID;
+            this.Op = op;
+            this.Logger = logger;
+        }
+
+        internal ILogger Logger { get; private set; }
+
+        public string TransType { get; set; }
+
+        public string Gid { get; set; }
+
+        public string BranchID { get; set; }
+
+        public string Op { get; set; }
+
+        public int BarrierID { get; set; }
+
+        public async Task Call(DbConnection db, Func<DbTransaction, Task> busiCall)
+        {
+            this.BarrierID = this.BarrierID + 1;
+            var bid = this.BarrierID.ToString().PadLeft(2, '0');
+
+            // check the connection state
+            if(db.State != System.Data.ConnectionState.Open) await db.OpenAsync();
+
+            var tx = db.BeginTransaction();
+
+            try
+            {
+                var originType = BarrierStatic.TypeDict.TryGetValue(this.Op, out var ot) ? ot : string.Empty;
+                
+                var originAffected = await DbUtils.InsertBarrier(db, this.TransType, this.Gid, this.BranchID, originType, bid, this.Op, tx);
+                var currentAffected = await DbUtils.InsertBarrier(db, this.TransType, this.Gid, this.BranchID, this.Op, bid, this.Op, tx);
+
+                Logger?.LogDebug("originAffected: {originAffected} currentAffected: {currentAffected}", originAffected, currentAffected);
+
+                var isNullCompensation = IsNullCompensation(this.Op, originAffected);
+                var isDuplicateOrPend = IsDuplicateOrPend(currentAffected);
+
+                if (isNullCompensation || isDuplicateOrPend)
+                {
+                    Logger?.LogInformation("Will not exec busiCall, isNullCompensation={isNullCompensation}, isDuplicateOrPend={isDuplicateOrPend}", isNullCompensation, isDuplicateOrPend);
+                    tx.Commit();
+                    return;
+                }
+
+                await busiCall.Invoke(tx);
+                tx.Commit();
+            }
+            catch (Exception ex)
+            {
+                Logger?.LogError(ex, "Call exception");
+                tx.Rollback();
+            }
+        }
+
+        public async Task<string> QueryPrepared(DbConnection db)
+        {
+            bool isErr = false;
+            try
+            {
+                var tmp = DbUtils.InsertBarrier(db, this.TransType, this.Gid, "00", "msg", "01", "rollback");
+            }
+            catch (Exception ex)
+            {
+                Logger?.LogInformation(ex, "QueryPrepared error");
+                isErr = true;
+            }
+
+            var reason = string.Empty;
+
+            if (!isErr)
+            {
+                var sql = string.Format("select reason from {0} where gid=@gid and branch_id=@branch_id and op=@op and barrier_id=@barrier_id", Constant.Barrier.TABLE_NAME);
+
+                reason = await db.QueryFirstOrDefaultAsync(
+                    sql,
+                    new { gid = this.Gid, branch_id = this.BranchID, op = this.Op, barrier_id = this.BarrierID });
+            }
+
+            if (reason.Equals("rollback")) return "FAILURE";
+
+            return string.Empty;
+        }
+
+        /// <summary>
+        /// 空补偿
+        /// </summary>
+        /// <param name="op"></param>
+        /// <param name="originAffected"></param>
+        /// <returns></returns>
+        private bool IsNullCompensation(string op, int originAffected)
+            => (op.Equals(Constant.BranchCancel) || op.Equals(Constant.Request.BRANCH_COMPENSATE)) && originAffected > 0;
+
+        /// <summary>
+        /// 这个是重复请求或者悬挂
+        /// </summary>
+        /// <param name="currentAffected"></param>
+        /// <returns></returns>
+        private bool IsDuplicateOrPend(int currentAffected)
+            => currentAffected == 0;
+
+        public bool IsInValid()
+        {
+            return string.IsNullOrWhiteSpace(this.TransType)
+                || string.IsNullOrWhiteSpace(this.Gid)
+                || string.IsNullOrWhiteSpace(this.BranchID)
+                || string.IsNullOrWhiteSpace(this.Op);
+        }
+
+        public override string ToString()
+            => $"transInfo: {TransType} {Gid} {BranchID} {Op}";
+    }
+
+    internal class BarrierStatic
+    {
+        internal static readonly Dictionary<string, string> TypeDict = new Dictionary<string, string>()
+        {
+            { Constant.BranchCancel, Constant.BranchTry },
+            { Constant.Request.BRANCH_COMPENSATE, Constant.Request.BRANCH_ACTION },
+        };
+    }
+}

src/Dtmcli/Barrier/DbUtils.cs

@@ -0,0 +1,28 @@
+using Dapper;
+using Dtmcli.DtmImp;
+using System.Data.Common;
+using System.Threading.Tasks;
+
+namespace Dtmcli
+{
+    public static class DbUtils
+    {
+        public static async Task<int> InsertBarrier(DbConnection db, string transType, string gid, string branchID, string op, string barrierID, string reason, DbTransaction tx = null)
+        {
+            if (db == null) return -1;
+            if (string.IsNullOrWhiteSpace(op)) return 0;
+
+            var str = string.Concat(Constant.Barrier.TABLE_NAME, "(trans_type, gid, branch_id, op, barrier_id, reason) values(@trans_type,@gid,@branch_id,@op,@barrier_id,@reason)");
+            var sql = DbSpecialDelegate.Instance.GetDBSpecial().GetInsertIgnoreTemplate(str, Constant.Barrier.PG_CONSTRAINT);
+
+            sql = DbSpecialDelegate.Instance.GetDBSpecial().GetPlaceHoldSQL(sql);
+
+            var result = await db.ExecuteAsync(
+                sql, 
+                new { trans_type = transType, gid = gid, branch_id = branchID, op = op, barrier_id = barrierID, reason = reason },
+                transaction: tx);
+
+            return result;
+        }
+    }
+}

src/Dtmcli/Barrier/DefaultBranchBarrierFactory.cs

@@ -0,0 +1,26 @@
+using Microsoft.Extensions.Logging;
+using System;
+
+namespace Dtmcli
+{
+    public class DefaultBranchBarrierFactory : IBranchBarrierFactory
+    {
+        private readonly ILogger _logger;
+
+        public DefaultBranchBarrierFactory(ILoggerFactory loggerFactory)
+        { 
+            this._logger = loggerFactory.CreateLogger<DefaultBranchBarrierFactory>();
+        }
+
+        public BranchBarrier CreateBranchBarrier(string transType, string gid, string branchID, string op, ILogger logger = null)
+        {
+            if(logger == null) logger = _logger;
+
+            var ti = new BranchBarrier(transType, gid, branchID, op, logger);
+
+            if (ti.IsInValid()) throw new Exception($"invalid trans info: {ti.ToString()}");
+
+            return ti;
+        }
+    }
+}

src/Dtmcli/Barrier/IBranchBarrierFactory.cs

@@ -0,0 +1,9 @@
+using Microsoft.Extensions.Logging;
+
+namespace Dtmcli
+{
+    public interface IBranchBarrierFactory
+    {
+        BranchBarrier CreateBranchBarrier(string transType, string gid, string branchID, string op, ILogger logger = null);
+    }
+}

src/Dtmcli/Constant.cs

@@ -103,5 +103,16 @@ internal class Request
 
             internal static readonly string TYPE_SAGA = "saga";
         }
+
+        internal class Barrier
+        {
+            internal static readonly string TABLE_NAME = "dtm_barrier.barrier";
+
+            internal static readonly string DBTYPE_MYSQL = "mysql";
+
+            internal static readonly string DBTYPE_POSTGRES = "postgres";
+
+            internal static readonly string PG_CONSTRAINT = "uniq_barrier";
+        }
     }
 }