MongoDB module ACID transaction implementation

framework/src/Volo.Abp.MongoDB/Volo.Abp.MongoDB.csproj

@@ -15,7 +15,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="MongoDB.Driver" Version="2.10.4" />
+    <PackageReference Include="MongoDB.Driver" Version="2.11.0" />
   </ItemGroup>
 
   <ItemGroup>

framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs

@@ -29,6 +29,8 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
 
         public virtual IMongoDatabase Database => DbContext.Database;
 
+        public virtual IClientSessionHandle SessionHandle => DbContext.SessionHandle;
+
         public virtual TMongoDbContext DbContext => DbContextProvider.GetDbContext();
 
         protected IMongoDbContextProvider<TMongoDbContext> DbContextProvider { get; }
@@ -60,10 +62,21 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
         {
             await ApplyAbpConceptsForAddedEntityAsync(entity);
 
-            await Collection.InsertOneAsync(
+            if (SessionHandle != null)
-                entity,
+            {
-                cancellationToken: GetCancellationToken(cancellationToken)
+                await Collection.InsertOneAsync(
-            );
+                    SessionHandle,
+                    entity,
+                    cancellationToken: GetCancellationToken(cancellationToken)
+                );
+            }
+            else
+            {
+                await Collection.InsertOneAsync(
+                    entity,
+                    cancellationToken: GetCancellationToken(cancellationToken)
+                );
+            }
 
             return entity;
         }
@@ -88,12 +101,25 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             await TriggerDomainEventsAsync(entity);
 
             var oldConcurrencyStamp = SetNewConcurrencyStamp(entity);
+            ReplaceOneResult result;
 
-            var result = await Collection.ReplaceOneAsync(
+            if (SessionHandle != null)
-                CreateEntityFilter(entity, true, oldConcurrencyStamp),
+            {
-                entity,
+                result = await Collection.ReplaceOneAsync(
-                cancellationToken: GetCancellationToken(cancellationToken)
+                    SessionHandle,
-            );
+                    CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                    entity,
+                    cancellationToken: GetCancellationToken(cancellationToken)
+                );
+            }
+            else
+            {
+                result = await Collection.ReplaceOneAsync(
+                    CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                    entity,
+                    cancellationToken: GetCancellationToken(cancellationToken)
+                );
+            }
 
             if (result.MatchedCount <= 0)
             {
@@ -114,11 +140,25 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             if (entity is ISoftDelete softDeleteEntity && !IsHardDeleted(entity))
             {
                 softDeleteEntity.IsDeleted = true;
-                var result = await Collection.ReplaceOneAsync(
+                ReplaceOneResult result;
-                    CreateEntityFilter(entity, true, oldConcurrencyStamp),
+
-                    entity,
+                if (SessionHandle != null)
-                    cancellationToken: GetCancellationToken(cancellationToken)
+                {
-                );
+                    result = await Collection.ReplaceOneAsync(
+                        SessionHandle,
+                        CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                        entity,
+                        cancellationToken: GetCancellationToken(cancellationToken)
+                    );
+                }
+                else
+                {
+                    result = await Collection.ReplaceOneAsync(
+                        CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                        entity,
+                        cancellationToken: GetCancellationToken(cancellationToken)
+                    );
+                }
 
                 if (result.MatchedCount <= 0)
                 {
@@ -127,10 +167,23 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             }
             else
             {
-                var result = await Collection.DeleteOneAsync(
+                DeleteResult result;
-                    CreateEntityFilter(entity, true, oldConcurrencyStamp),
+
-                    GetCancellationToken(cancellationToken)
+                if (SessionHandle != null)
-                );
+                {
+                    result = await Collection.DeleteOneAsync(
+                        SessionHandle,
+                        CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                        cancellationToken: GetCancellationToken(cancellationToken)
+                    );
+                }
+                else
+                {
+                    result = await Collection.DeleteOneAsync(
+                        CreateEntityFilter(entity, true, oldConcurrencyStamp),
+                        GetCancellationToken(cancellationToken)
+                    );
+                }
 
                 if (result.DeletedCount <= 0)
                 {
@@ -404,6 +457,13 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             bool includeDetails = true,
             CancellationToken cancellationToken = default)
         {
+            if (SessionHandle != null)
+            {
+                return await Collection
+                    .Find(SessionHandle, RepositoryFilterer.CreateEntityFilter(id, true))
+                    .FirstOrDefaultAsync(GetCancellationToken(cancellationToken));
+            }
+
             return await Collection
                 .Find(RepositoryFilterer.CreateEntityFilter(id, true))
                 .FirstOrDefaultAsync(GetCancellationToken(cancellationToken));

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/AbpMongoDbContext.cs

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using MongoDB.Bson;
 using MongoDB.Driver;
 using Volo.Abp.DependencyInjection;
 
@@ -10,18 +11,23 @@ namespace Volo.Abp.MongoDB
 
         public IMongoDatabase Database { get; private set; }
 
+        public IClientSessionHandle SessionHandle { get; private set; }
+
         protected internal virtual void CreateModel(IMongoModelBuilder modelBuilder)
         {
 
         }
 
-        public virtual void InitializeDatabase(IMongoDatabase database)
+        public virtual void InitializeDatabase(IMongoDatabase database, IClientSessionHandle sessionHandle)
         {
             Database = database;
+            SessionHandle = sessionHandle;
         }
 
         public virtual IMongoCollection<T> Collection<T>()
         {
+            CreateCollectionIfNotExists<T>();
+
             return Database.GetCollection<T>(GetCollectionName<T>());
         }
 
@@ -41,5 +47,23 @@ namespace Volo.Abp.MongoDB
 
             return model;
         }
+
+        protected virtual void CreateCollectionIfNotExists<T>()
+        {
+            var collectionName = GetCollectionName<T>();
+
+            if (!CollectionExists(collectionName))
+            {
+                Database.CreateCollection(collectionName);
+            }
+        }
+
+        protected virtual bool CollectionExists(string collectionName)
+        {
+            var filter = new BsonDocument("name", collectionName);
+            var options = new ListCollectionNamesOptions { Filter = filter };
+
+            return Database.ListCollectionNames(options).Any();
+        }
     }
-}
+}

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/IAbpMongoDbContext.cs

@@ -7,5 +7,7 @@ namespace Volo.Abp.MongoDB
         IMongoDatabase Database { get; }
 
         IMongoCollection<T> Collection<T>();
+
+        IClientSessionHandle SessionHandle { get; }
     }
-}
+}

framework/src/Volo.Abp.MongoDB/Volo/Abp/Uow/MongoDB/MongoDbTransactionApi.cs

@@ -0,0 +1,42 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+
+namespace Volo.Abp.Uow.MongoDB
+{
+    public class MongoDbTransactionApi : ITransactionApi, ISupportsRollback
+    {
+        public IClientSessionHandle SessionHandle { get; }
+
+        public MongoDbTransactionApi(IClientSessionHandle sessionHandle)
+        {
+            SessionHandle = sessionHandle;
+        }
+
+        public async Task CommitAsync()
+        {
+            await SessionHandle.CommitTransactionAsync();
+        }
+
+        protected void Commit()
+        {
+            SessionHandle.CommitTransaction();
+        }
+
+        public void Dispose()
+        {
+            SessionHandle.Dispose();
+        }
+
+        public void Rollback()
+        {
+            SessionHandle.AbortTransaction();
+        }
+
+        public async Task RollbackAsync(CancellationToken cancellationToken)
+        {
+            await SessionHandle.AbortTransactionAsync(cancellationToken);
+        }
+    }
+}

framework/src/Volo.Abp.MongoDB/Volo/Abp/Uow/MongoDB/UnitOfWorkMongoDbContextProvider.cs

@@ -1,5 +1,6 @@
 using System;
 using Microsoft.Extensions.DependencyInjection;
+using MongoDB.Bson;
 using MongoDB.Driver;
 using Volo.Abp.Data;
 using Volo.Abp.MongoDB;
@@ -25,7 +26,8 @@ namespace Volo.Abp.Uow.MongoDB
             var unitOfWork = _unitOfWorkManager.Current;
             if (unitOfWork == null)
             {
-                throw new AbpException($"A {nameof(IMongoDatabase)} instance can only be created inside a unit of work!");
+                throw new AbpException(
+                    $"A {nameof(IMongoDatabase)} instance can only be created inside a unit of work!");
             }
 
             var connectionString = _connectionStringResolver.Resolve<TMongoDbContext>();
@@ -41,18 +43,42 @@ namespace Volo.Abp.Uow.MongoDB
             //TODO: Create only single MongoDbClient per connection string in an application (extract MongoClientCache for example).
             var databaseApi = unitOfWork.GetOrAddDatabaseApi(
                 dbContextKey,
-                () =>
+                () => new MongoDbDatabaseApi<TMongoDbContext>(CreateDbContext(unitOfWork,mongoUrl,databaseName)));
+
+            return ((MongoDbDatabaseApi<TMongoDbContext>) databaseApi).DbContext;
+        }
+
+        private TMongoDbContext CreateDbContext(IUnitOfWork unitOfWork, MongoUrl mongoUrl, string databaseName)
+        {
+            var client = new MongoClient(mongoUrl);
+            var database = client.GetDatabase(databaseName);
+            var dbContext = unitOfWork.ServiceProvider.GetRequiredService<TMongoDbContext>();
+
+            if (unitOfWork.Options.IsTransactional)
+            {
+                var session = client.StartSession();
+
+                if (unitOfWork.Options.Timeout.HasValue)
                 {
-                    var database = new MongoClient(mongoUrl).GetDatabase(databaseName);
+                    session.AdvanceOperationTime(new BsonTimestamp(unitOfWork.Options.Timeout.Value));
+                }
 
-                    var dbContext = unitOfWork.ServiceProvider.GetRequiredService<TMongoDbContext>();
+                session.StartTransaction();
 
-                    dbContext.ToAbpMongoDbContext().InitializeDatabase(database);
+                var transactionApiKey = $"MongoDb_{mongoUrl}";
+                unitOfWork.AddTransactionApi(
+                    transactionApiKey,
+                    new MongoDbTransactionApi(session)
+                );
 
-                    return new MongoDbDatabaseApi<TMongoDbContext>(dbContext);
+                dbContext.ToAbpMongoDbContext().InitializeDatabase(database, session);
-                });
+            }
+            else
+            {
+                dbContext.ToAbpMongoDbContext().InitializeDatabase(database, null);
+            }
 
-            return ((MongoDbDatabaseApi<TMongoDbContext>)databaseApi).DbContext;
+            return dbContext;
         }
     }
-}
+}

framework/test/Volo.Abp.MongoDB.Tests/Volo/Abp/MongoDB/AbpMongoDbTestModule.cs

@@ -16,9 +16,11 @@ namespace Volo.Abp.MongoDB
     {
         public override void ConfigureServices(ServiceConfigurationContext context)
         {
-            var connectionString = MongoDbFixture.ConnectionString.EnsureEndsWith('/') +
+            var stringArray = MongoDbFixture.ConnectionString.Split('?');
+
+            var connectionString = stringArray[0].EnsureEndsWith('/') +
                                    "Db_" +
-                                   Guid.NewGuid().ToString("N");
+                                   Guid.NewGuid().ToString("N") + "/?" + stringArray[1];
 
             Configure<AbpDbConnectionOptions>(options =>
             {

framework/test/Volo.Abp.MongoDB.Tests/Volo/Abp/MongoDB/MongoClientExtension.cs

@@ -0,0 +1,84 @@
+using System;
+using System.Threading;
+using MongoDB.Driver;
+
+namespace Volo.Abp.MongoDB
+{
+    /// <summary>
+    /// TODO It can be removed, when Mongo2Go solves this issue : https://github.com/Mongo2Go/Mongo2Go/issues/89
+    /// </summary>
+     internal static class MongoClientExtension
+    {
+        private static readonly TimeSpan InitialDelay = TimeSpan.FromMilliseconds(500);
+        private static readonly TimeSpan MaxDelay = TimeSpan.FromSeconds(5000);
+
+
+        public static void EnsureReplicationSetReady(this IMongoClient mongoClient)
+        {
+            var delay = InitialDelay;
+            var database = mongoClient.GetDatabase("__dummy-db");
+            try
+            {
+                while (true)
+                {
+                    try
+                    {
+                        _ = database.GetCollection<DummyEntry>("__dummy");
+                        database.DropCollection("__dummy");
+
+                        var session = mongoClient.StartSession();
+
+                        try
+                        {
+                            session.StartTransaction();
+                            session.AbortTransaction();
+                        }
+                        finally
+                        {
+                            session.Dispose();
+                        }
+                        break;
+                    }
+                    catch (NotSupportedException) { }
+
+                    Thread.Sleep(delay);
+                    delay = Min(Double(delay), MaxDelay);
+                }
+            }
+            finally
+            {
+                mongoClient.DropDatabase("__dummy-db");
+            }
+        }
+
+        private static TimeSpan Min(TimeSpan left, TimeSpan right)
+        {
+            return new TimeSpan(Math.Min(left.Ticks, right.Ticks));
+        }
+
+        private static TimeSpan Double(TimeSpan timeSpan)
+        {
+            long ticks;
+            try
+            {
+                ticks = checked(timeSpan.Ticks * 2);
+            }
+            catch (OverflowException)
+            {
+                if (timeSpan.Ticks >= 0)
+                {
+                    return TimeSpan.MaxValue;
+                }
+
+                return TimeSpan.MinValue;
+            }
+
+            return new TimeSpan(ticks);
+        }
+
+        private sealed class DummyEntry
+        {
+            public int Id { get; set; }
+        }
+    }
+}

framework/test/Volo.Abp.MongoDB.Tests/Volo/Abp/MongoDB/MongoDbFixture.cs

@@ -1,5 +1,6 @@
 using System;
 using Mongo2Go;
+using MongoDB.Driver;
 
 namespace Volo.Abp.MongoDB
 {
@@ -10,8 +11,12 @@ namespace Volo.Abp.MongoDB
 
         static MongoDbFixture()
         {
-            MongoDbRunner = MongoDbRunner.Start();
+            MongoDbRunner = MongoDbRunner.Start(singleNodeReplSet: true, singleNodeReplSetWaitTimeout: 10);
             ConnectionString = MongoDbRunner.ConnectionString;
+
+            //TODO It can be removed, when Mongo2Go solves this issue : https://github.com/Mongo2Go/Mongo2Go/issues/89
+            var client = new MongoClient(MongoDbRunner.ConnectionString);
+            client.EnsureReplicationSetReady();
         }
 
         public void Dispose()

framework/test/Volo.Abp.MongoDB.Tests/Volo/Abp/MongoDB/Transactions/Transaction_Tests.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Shouldly;
+using Volo.Abp.Domain.Repositories;
+using Volo.Abp.TestApp.Domain;
+using Volo.Abp.TestApp.Testing;
+using Volo.Abp.Uow;
+using Xunit;
+
+namespace Volo.Abp.MongoDB.Transactions
+{
+    public class Transaction_Tests : TestAppTestBase<AbpMongoDbTestModule>
+    {
+
+        private readonly IBasicRepository<Person, Guid> _personRepository;
+        private readonly IBasicRepository<City, Guid> _cityRepository;
+        private readonly IUnitOfWorkManager _unitOfWorkManager;
+
+        public Transaction_Tests()
+        {
+            _personRepository = ServiceProvider.GetRequiredService<IBasicRepository<Person, Guid>>();
+            _cityRepository = ServiceProvider.GetRequiredService<IBasicRepository<City, Guid>>();
+            _unitOfWorkManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
+        }
+
+        [Fact]
+        public async Task Should_Rollback_Transaction_When_An_Exception_Is_Thrown()
+        {
+            var personId = Guid.NewGuid();
+            const string exceptionMessage = "thrown to rollback the transaction!";
+
+            try
+            {
+                await WithUnitOfWorkAsync(new AbpUnitOfWorkOptions {IsTransactional = true}, async () =>
+                {
+                    await _personRepository.InsertAsync(new Person(personId, "Adam", 42));
+                    throw new Exception(exceptionMessage);
+                });
+            }
+            catch (Exception e) when (e.Message == exceptionMessage)
+            {
+
+            }
+
+            var person = await _personRepository.FindAsync(personId);
+            person.ShouldBeNull();
+        }
+
+        [Fact]
+        public async Task Should_Rollback_Transaction_Manually()
+        {
+            var personId = Guid.NewGuid();
+
+            await WithUnitOfWorkAsync(new AbpUnitOfWorkOptions {IsTransactional = true}, async () =>
+            {
+                _unitOfWorkManager.Current.ShouldNotBeNull();
+
+                await _personRepository.InsertAsync(new Person(personId, "Adam", 42));
+
+                await _unitOfWorkManager.Current.RollbackAsync();
+            });
+
+            var person = await _personRepository.FindAsync(personId);
+            person.ShouldBeNull();
+        }
+
+        [Fact]
+        public async Task Should_Rollback_Transaction_Manually_With_Double_DbContext_Transaction()
+        {
+            var personId = Guid.NewGuid();
+            var cityId = Guid.NewGuid();
+
+            using (var scope = ServiceProvider.CreateScope())
+            {
+                var uowManager = scope.ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
+
+                using (uowManager.Begin(new AbpUnitOfWorkOptions {IsTransactional = true}))
+                {
+                    _unitOfWorkManager.Current.ShouldNotBeNull();
+
+                    await _personRepository.InsertAsync(new Person(personId, "Adam", 42));
+                    await _cityRepository.InsertAsync(new City(cityId, cityId.ToString()));
+
+                    await _unitOfWorkManager.Current.SaveChangesAsync();
+
+                    //Will automatically rollback since not called the Complete!
+                }
+            }
+
+            (await _personRepository.FindAsync(personId)).ShouldBeNull();
+            (await _cityRepository.FindAsync(cityId)).ShouldBeNull();
+        }
+    }
+}