msg support add topic (#64)

* feat(msg): add topic

* add test and sample for msg topic

* test: grpc msg topic

Author: catcherwong

samples/DtmSample/Controllers/MsgTestController.cs

@@ -218,5 +218,47 @@ public async Task<IActionResult> MsgMongoQueryPrepared(CancellationToken cancell
             var res = await bb.MongoQueryPrepared(cli);
             return Ok(new { dtm_result = res });
         }
+
+        /// <summary>
+        /// MSG with not exist topic will get 【topic not found】
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        [HttpGet("msg-topic-notfound")]
+        public async Task<IActionResult> MsgWithTopicNotFound(CancellationToken cancellationToken)
+        {
+            var gid = await _dtmClient.GenGid(cancellationToken);
+            var req = new TransRequest("1", -30);
+            var msg = _transFactory.NewMsg(gid)
+                .AddTopic("not_exist_topic", req);
+
+            await msg.Prepare(_settings.BusiUrl + "/msg-queryprepared", cancellationToken);
+            await msg.Submit(cancellationToken);
+
+            return Ok(TransResponse.BuildSucceedResponse());
+        }
+
+        /// <summary>
+        /// MSG with exist topic
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        [HttpGet("msg-topic")]
+        public async Task<IActionResult> MsgWithTopic(CancellationToken cancellationToken)
+        {
+            var gid = await _dtmClient.GenGid(cancellationToken);
+
+            // should subscribe at first
+            var topic ="mytopic";
+
+            var req = new TransRequest("1", -30);
+            var msg = _transFactory.NewMsg(gid)
+                .AddTopic(topic, req);
+
+            await msg.Prepare(_settings.BusiUrl + "/msg-queryprepared", cancellationToken);
+            await msg.Submit(cancellationToken);
+
+            return Ok(TransResponse.BuildSucceedResponse());
+        }
     }
 }

src/DtmCommon/Constant.cs

@@ -18,6 +18,8 @@ public class Constant
         /// </summary>
         public static readonly string ResultDuplicated = "DUPLICATED";
 
+        public static readonly string MsgTopicPrefix = "topic://";
+
         internal class Op
         {
             internal static readonly string Submit = "Submit";

src/Dtmcli/Msg/Msg.cs

@@ -34,6 +34,11 @@ public Msg Add(string action, object postData)
             return this;
         }
 
+        public Msg AddTopic(string topic, object postData)
+        {
+            return this.Add($"{DtmCommon.Constant.MsgTopicPrefix}{topic}", postData);
+        }
+
         public async Task Prepare(string queryPrepared, CancellationToken cancellationToken = default)
         {
             this._transBase.QueryPrepared = !string.IsNullOrWhiteSpace(queryPrepared)? queryPrepared : this._transBase.QueryPrepared;

src/Dtmgrpc/Msg/MsgGrpc.cs

@@ -38,6 +38,11 @@ public MsgGrpc Add(string action, IMessage payload)
             return this;
         }
 
+        public MsgGrpc AddTopic(string topic, IMessage payload)
+        {
+            return this.Add($"{Constant.MsgTopicPrefix}{topic}", payload);
+        }
+
         public async Task Prepare(string queryPrepared, CancellationToken cancellationToken = default)
         {
             this._transBase.QueryPrepared = !string.IsNullOrWhiteSpace(queryPrepared) ? queryPrepared : this._transBase.QueryPrepared;

src/Dtmgrpc/dtmgpb/dtmgimp.proto

@@ -13,6 +13,9 @@ service Dtm {
   rpc Prepare(DtmRequest) returns (google.protobuf.Empty) {}
   rpc Abort(DtmRequest) returns (google.protobuf.Empty) {}
   rpc RegisterBranch(DtmBranchRequest) returns (google.protobuf.Empty) {}
+  rpc Subscribe(DtmTopicRequest) returns (google.protobuf.Empty){}
+  rpc Unsubscribe(DtmTopicRequest) returns (google.protobuf.Empty){}
+  rpc DeleteTopic(DtmTopicRequest) returns (google.protobuf.Empty){}
 }
 
 message DtmTransOptions {
@@ -50,3 +53,9 @@ message DtmBranchRequest {
   map<string, string> Data = 5;
   bytes BusiPayload = 6;
 }
+
+message DtmTopicRequest {
+  string Topic = 1;
+  string URL = 2;
+  string Remark = 3;
+}

tests/Dtmcli.Tests/MsgTests.cs

@@ -5,6 +5,7 @@
 using System;
 using System.Collections.Generic;
 using System.Data.Common;
+using System.Linq;
 using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
@@ -45,6 +46,7 @@ public async void Submit_Should_Succeed()
 
             msg.Add(busi + "/TransOut", req)
                .Add(busi + "/TransIn", req)
+               .AddTopic("test-topic", req)
                .EnableWaitResult()
                .SetRetryInterval(10)
                .SetTimeoutToFail(100)
@@ -209,10 +211,11 @@ protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage
                 Assert.Equal(100, transBase.TimeoutToFail);
                 Assert.Contains("bh1", transBase.BranchHeaders.Keys);
                 Assert.Contains("bh2", transBase.BranchHeaders.Keys);
-                Assert.Equal(2, transBase.Payloads.Count);
-                Assert.Equal(2, transBase.Steps.Count);
+                Assert.Equal(3, transBase.Payloads.Count);
+                Assert.Equal(3, transBase.Steps.Count);
+                Assert.Contains("topic://test-topic", transBase.Steps.SelectMany(x => x.Values).ToList());
 
-                if(request.RequestUri.AbsolutePath.Contains("submit",StringComparison.OrdinalIgnoreCase))
+                if (request.RequestUri.AbsolutePath.Contains("submit",StringComparison.OrdinalIgnoreCase))
                 {
                     Assert.NotEmpty(transBase.CustomData);
                     Assert.Contains("10", transBase.CustomData);

tests/Dtmgrpc.Tests/MsgTests.cs

@@ -48,6 +48,7 @@ public async void Submit_Should_Succeed()
 
             msg.Add(busi + "/TransOut", req)
                .Add(busi + "/TransIn", req)
+               .AddTopic("mytopic", req)
                .EnableWaitResult()
                .SetRetryInterval(10)
                .SetTimeoutToFail(100)