Refactor mongodb module for new transaction mode

samples/Sample.RabbitMQ.MongoDB/Controllers/ValuesController.cs

@@ -22,10 +22,13 @@ namespace Sample.RabbitMQ.MongoDB.Controllers
         [Route("~/publish")]
         public IActionResult PublishWithTrans()
         {
+            //var mycollection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
+            //mycollection.InsertOne(new BsonDocument { { "test", "test" } });
+
             using (var session = _client.StartSession())
             using (var trans = _capPublisher.CapTransaction.Begin(session))
             {
-                var collection = _client.GetDatabase("TEST").GetCollection<BsonDocument>("test");
+                var collection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
                 collection.InsertOne(session, new BsonDocument { { "hello", "world" } });
 
                 _capPublisher.Publish("sample.rabbitmq.mongodb", DateTime.Now);
@@ -39,17 +42,17 @@ namespace Sample.RabbitMQ.MongoDB.Controllers
         public IActionResult PublishNotAutoCommit()
         {
             using (var session = _client.StartSession())
-            using (_capPublisher.CapTransaction.Begin(session,true))
+            using (_capPublisher.CapTransaction.Begin(session, true))
             {
-                var collection = _client.GetDatabase("TEST").GetCollection<BsonDocument>("test");
-                collection.InsertOne(session, new BsonDocument { { "hello", "world" } });
+                var collection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
+                collection.InsertOne(session, new BsonDocument { { "hello2", "world2" } });
 
                 _capPublisher.Publish("sample.rabbitmq.mongodb", DateTime.Now);
             }
 
             return Ok();
         }
-         
+
         [Route("~/publish/without/trans")]
         public IActionResult PublishWithoutTrans()
         {

samples/Sample.RabbitMQ.MongoDB/Startup.cs

@@ -1,5 +1,4 @@
-using DotNetCore.CAP;
-using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Configuration;
@@ -19,21 +18,11 @@ namespace Sample.RabbitMQ.MongoDB
 
         public void ConfigureServices(IServiceCollection services)
         {
-            services.AddSingleton<IMongoClient>(new MongoClient(Configuration.GetConnectionString("MongoDB")));
+            services.AddSingleton<IMongoClient>(new MongoClient("mongodb://192.168.10.110:27017,192.168.10.110:27018,192.168.10.110:27019/?replicaSet=rs0"));
             services.AddCap(x =>
             {
-                x.UseMongoDB();
-
-                var mq = new RabbitMQOptions();
-                Configuration.GetSection("RabbitMQ").Bind(mq);
-                x.UseRabbitMQ(cfg =>
-                {
-                    cfg.HostName = mq.HostName;
-                    cfg.Port = mq.Port;
-                    cfg.UserName = mq.UserName;
-                    cfg.Password = mq.Password;
-                });
-
+                x.UseMongoDB("mongodb://192.168.10.110:27017,192.168.10.110:27018,192.168.10.110:27019/?replicaSet=rs0");
+                x.UseRabbitMQ("localhost");
                 x.UseDashboard();
             });
             services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_1);
@@ -46,9 +35,7 @@ namespace Sample.RabbitMQ.MongoDB
                 app.UseDeveloperExceptionPage();
             }
 
-            app.UseMvc();
-
-            app.UseCap();
+            app.UseMvc(); 
         }
     }
 }

test/DotNetCore.CAP.MongoDB.Test/MongoDBMonitoringApiTest.cs

@@ -17,20 +17,19 @@ namespace DotNetCore.CAP.MongoDB.Test
         {
             _api = new MongoDBMonitoringApi(MongoClient, MongoDBOptions);
 
-            var helper = new MongoDBUtil();
             var collection = Database.GetCollection<CapPublishedMessage>(MongoDBOptions.PublishedCollection);
             collection.InsertMany(new[]
             {
                 new CapPublishedMessage
                 {
-                    Id = helper.GetNextSequenceValue(Database,MongoDBOptions.PublishedCollection),
+                    Id = SnowflakeId.Default().NextId(),
                     Added = DateTime.Now.AddHours(-1),
                     StatusName = "Failed",
                     Content = "abc"
                 },
                 new CapPublishedMessage
                 {
-                    Id = helper.GetNextSequenceValue(Database,MongoDBOptions.PublishedCollection),
+                    Id =  SnowflakeId.Default().NextId(),
                     Added = DateTime.Now,
                     StatusName = "Failed",
                     Content = "bbc"

test/DotNetCore.CAP.MongoDB.Test/MongoDBStorageConnectionTest.cs

@@ -11,20 +11,23 @@ namespace DotNetCore.CAP.MongoDB.Test
     [Collection("MongoDB")]
     public class MongoDBStorageConnectionTest : DatabaseTestHost
     {
-        private IStorageConnection _connection => 
+        private IStorageConnection _connection =>
             Provider.GetService<MongoDBStorage>().GetConnection();
 
         [Fact]
-        public async void StoreReceivedMessageAsync_TestAsync()
+        public void StoreReceivedMessageAsync_TestAsync()
         {
-            var id = await _connection.StoreReceivedMessageAsync(new CapReceivedMessage(new MessageContext
+            var messageContext = new MessageContext
             {
                 Group = "test",
                 Name = "test",
                 Content = "test-content"
-            }));
+            };
 
-            id.Should().BeGreaterThan(0);
+            _connection.StoreReceivedMessage(new CapReceivedMessage(messageContext)
+            {
+                Id = SnowflakeId.Default().NextId()
+            });
         }
 
         [Fact]
@@ -45,14 +48,17 @@ namespace DotNetCore.CAP.MongoDB.Test
 
             msgs.Should().BeEmpty();
 
+            var id = SnowflakeId.Default().NextId();
+
             var msg = new CapReceivedMessage
             {
+                Id = id,
                 Group = "test",
                 Name = "test",
                 Content = "test-content",
                 StatusName = StatusName.Failed
             };
-            var id = await _connection.StoreReceivedMessageAsync(msg);
+            _connection.StoreReceivedMessage(msg);
 
             var collection = Database.GetCollection<CapReceivedMessage>(MongoDBOptions.ReceivedCollection);
 

test/DotNetCore.CAP.MongoDB.Test/MongoDBStorageTest.cs

@@ -1,6 +1,4 @@
 using FluentAssertions;
-using Microsoft.Extensions.DependencyInjection;
-using MongoDB.Bson;
 using MongoDB.Driver;
 using Xunit;
 
@@ -12,18 +10,12 @@ namespace DotNetCore.CAP.MongoDB.Test
         [Fact]
         public void InitializeAsync_Test()
         {
-            var storage = Provider.GetService<MongoDBStorage>();
             var names = MongoClient.ListDatabaseNames()?.ToList();
             names.Should().Contain(MongoDBOptions.DatabaseName);
 
             var collections = Database.ListCollectionNames()?.ToList();
             collections.Should().Contain(MongoDBOptions.PublishedCollection);
             collections.Should().Contain(MongoDBOptions.ReceivedCollection);
-            collections.Should().Contain(MongoDBOptions.CounterCollection);
-
-            var collection = Database.GetCollection<BsonDocument>(MongoDBOptions.CounterCollection);
-            collection.CountDocuments(new BsonDocument { { "_id", MongoDBOptions.PublishedCollection } }).Should().Be(1);
-            collection.CountDocuments(new BsonDocument { { "_id", MongoDBOptions.ReceivedCollection } }).Should().Be(1);
         }
     }
 }

test/DotNetCore.CAP.MongoDB.Test/MongoDBUtilTest.cs

@@ -1,30 +0,0 @@
-using System.Collections.Concurrent;
-using System.Threading.Tasks;
-using FluentAssertions;
-using Xunit;
-
-namespace DotNetCore.CAP.MongoDB.Test
-{
-    [Collection("MongoDB")]
-    public class MongoDBUtilTest : DatabaseTestHost
-    {
-        [Fact]
-        public async void GetNextSequenceValueAsync_Test()
-        {
-            var id = await new MongoDBUtil().GetNextSequenceValueAsync(Database, MongoDBOptions.ReceivedCollection);
-            id.Should().BeGreaterThan(0);
-        }
-
-        [Fact]
-        public void GetNextSequenceValue_Concurrency_Test()
-        {
-            var dic = new ConcurrentDictionary<int, int>();
-            Parallel.For(0, 30, (x) =>
-             {
-                 var id = new MongoDBUtil().GetNextSequenceValue(Database, MongoDBOptions.ReceivedCollection);
-                 id.Should().BeGreaterThan(0);
-                 dic.TryAdd(id, x).Should().BeTrue(); //The id shouldn't be same.
-             });
-        }
-    }
-}