Merge pull request #7464 from abpframework/liangshiwei/mongo

Add base aggregation API to MongoDB repository

docs/en/MongoDB.md

@@ -254,7 +254,7 @@ public async override Task DeleteAsync(
 
 ### Access to the MongoDB API
 
-In most cases, you want to hide MongoDB APIs behind a repository (this is the main purpose of the repository). However, if you want to access the MongoDB API over the repository, you can use `GetDatabaseAsync()` or `GetCollectionAsync()` extension methods. Example:
+In most cases, you want to hide MongoDB APIs behind a repository (this is the main purpose of the repository). However, if you want to access the MongoDB API over the repository, you can use `GetDatabaseAsync()`, `GetCollectionAsync()` or `GetAggregateAsync()` extension methods. Example:
 
 ```csharp
 public class BookService
@@ -270,6 +270,7 @@ public class BookService
     {
         IMongoDatabase database = await _bookRepository.GetDatabaseAsync();
         IMongoCollection<Book> books = await _bookRepository.GetCollectionAsync();
+        IAggregateFluent<Book> bookAggregate = await _bookRepository.GetAggregateAsync();
     }
 }
 ```

docs/zh-Hans/MongoDB.md

@@ -211,7 +211,7 @@ public async override Task DeleteAsync(
 
 #### 访问MongoDB API
 
-大多数情况下,你想要将MongoDB API隐藏在仓储后面(这是仓储的主要目的).如果你想在仓储之上访问MongoDB API,你可以使用`GetDatabase()`或`GetCollection()`方法.例如:
+大多数情况下,你想要将MongoDB API隐藏在仓储后面(这是仓储的主要目的).如果你想在仓储之上访问MongoDB API,你可以使用`GetDatabaseAsync()`, `GetAggregateAsync()` 或`GetCollectionAsync()`方法.例如:
 
 ```csharp
 public class BookService
@@ -223,10 +223,11 @@ public class BookService
         _bookRepository = bookRepository;
     }
 
-    public void Foo()
+    public async Task FooAsync()
     {
-        IMongoDatabase database = _bookRepository.GetDatabase();
+        IMongoDatabase database = await _bookRepository.GetDatabaseAsync();
-        IMongoCollection<Book> books = _bookRepository.GetCollection();
+        IMongoCollection<Book> books = await _bookRepository.GetCollectionAsync();
+        IAggregateFluent<Book> bookAggregate = await _bookRepository.GetAggregateAsync();
     }
 }
 ```

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

@@ -24,6 +24,8 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
         IMongoQueryable<TEntity> GetMongoQueryable();
 
         Task<IMongoQueryable<TEntity>> GetMongoQueryableAsync(CancellationToken cancellationToken = default);
+
+        Task<IAggregateFluent<TEntity>> GetAggregateAsync(CancellationToken cancellationToken = default);
     }
 
     public interface IMongoDbRepository<TEntity, TKey> : IMongoDbRepository<TEntity>, IRepository<TEntity, TKey>

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

@@ -16,6 +16,7 @@ using Volo.Abp.EventBus.Distributed;
 using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MongoDB;
+using Volo.Abp.MultiTenancy;
 
 namespace Volo.Abp.Domain.Repositories.MongoDB
 {
@@ -486,6 +487,17 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             );
         }
 
+        public async Task<IAggregateFluent<TEntity>> GetAggregateAsync(CancellationToken cancellationToken = default)
+        {
+            var dbContext = await GetDbContextAsync(cancellationToken);
+            var collection = await GetCollectionAsync(cancellationToken);
+
+            return ApplyDataFilters(
+                dbContext.SessionHandle != null
+                    ? collection.Aggregate(dbContext.SessionHandle)
+                    : collection.Aggregate());
+        }
+
         protected virtual bool IsHardDeleted(TEntity entity)
         {
             var hardDeletedEntities = UnitOfWorkManager?.Current?.Items.GetOrDefault(UnitOfWorkItemNames.HardDeletedEntities) as HashSet<IEntity>;
@@ -634,6 +646,22 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
             throw new AbpDbConcurrencyException("Database operation expected to affect 1 row but actually affected 0 row. Data may have been modified or deleted since entities were loaded. This exception has been thrown on optimistic concurrency check.");
         }
 
+        protected virtual IAggregateFluent<TEntity> ApplyDataFilters(IAggregateFluent<TEntity> aggregate)
+        {
+            if (typeof(ISoftDelete).IsAssignableFrom(typeof(TEntity)) && DataFilter.IsEnabled<ISoftDelete>())
+            {
+                aggregate = aggregate.Match(e => ((ISoftDelete)e).IsDeleted == false);
+            }
+
+            if (typeof(IMultiTenant).IsAssignableFrom(typeof(TEntity)) && DataFilter.IsEnabled<IMultiTenant>())
+            {
+                var tenantId = CurrentTenant.Id;
+                aggregate = aggregate.Match(e => ((IMultiTenant)e).TenantId == tenantId);
+            }
+
+            return aggregate;
+        }
+
         [Obsolete("This method will be removed in future versions.")]
         public QueryableExecutionModel GetExecutionModel()
         {

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

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using MongoDB.Driver;
 using MongoDB.Driver.Linq;
@@ -16,10 +17,10 @@ namespace Volo.Abp.Domain.Repositories
             return repository.ToMongoDbRepository().Database;
         }
 
-        public static Task<IMongoDatabase> GetDatabaseAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository)
+        public static Task<IMongoDatabase> GetDatabaseAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository, CancellationToken cancellationToken = default)
             where TEntity : class, IEntity<TKey>
         {
-            return repository.ToMongoDbRepository().GetDatabaseAsync();
+            return repository.ToMongoDbRepository().GetDatabaseAsync(cancellationToken);
         }
 
         [Obsolete("Use GetCollectionAsync method.")]
@@ -29,10 +30,10 @@ namespace Volo.Abp.Domain.Repositories
             return repository.ToMongoDbRepository().Collection;
         }
 
-        public static Task<IMongoCollection<TEntity>> GetCollectionAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository)
+        public static Task<IMongoCollection<TEntity>> GetCollectionAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository, CancellationToken cancellationToken = default)
             where TEntity : class, IEntity<TKey>
         {
-            return repository.ToMongoDbRepository().GetCollectionAsync();
+            return repository.ToMongoDbRepository().GetCollectionAsync(cancellationToken);
         }
 
         [Obsolete("Use GetMongoQueryableAsync method.")]
@@ -42,10 +43,16 @@ namespace Volo.Abp.Domain.Repositories
             return repository.ToMongoDbRepository().GetMongoQueryable();
         }
 
-        public static Task<IMongoQueryable<TEntity>> GetMongoQueryableAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository)
+        public static Task<IMongoQueryable<TEntity>> GetMongoQueryableAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository, CancellationToken cancellationToken = default)
             where TEntity : class, IEntity<TKey>
         {
-            return repository.ToMongoDbRepository().GetMongoQueryableAsync();
+            return repository.ToMongoDbRepository().GetMongoQueryableAsync(cancellationToken);
+        }
+
+        public static Task<IAggregateFluent<TEntity>> GetAggregateAsync<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository, CancellationToken cancellationToken = default)
+            where TEntity : class, IEntity<TKey>
+        {
+            return repository.ToMongoDbRepository().GetAggregateAsync(cancellationToken);
         }
 
         public static IMongoDbRepository<TEntity, TKey> ToMongoDbRepository<TEntity, TKey>(this IBasicRepository<TEntity, TKey> repository)