src/Dpz.Core.MongodbAccess/RepositoryUnitOfWork.cs 网站源码

Root / src / Dpz.Core.MongodbAccess / RepositoryUnitOfWork.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using Dpz.Core.Entity.Base;
using MongoDB.Driver;

namespace Dpz.Core.MongodbAccess;

public sealed class RepositoryUnitOfWork<T>(MongodbAccess<T> access, IClientSessionHandle session)
    : IRepository<T>
    where T : IBaseEntity
{
    public IMongoCollection<T> Collection => access.Collection;

    public IQueryable<T> MongodbQueryable =>
        Database
            .GetCollection<T>(access.CollectionName)
            .AsQueryable(session, new AggregateOptions { AllowDiskUse = true });

    public IQueryable<T> SearchFor(Expression<Func<T, bool>> predicate)
    {
        return MongodbQueryable.Where(predicate);
    }

    public IFindFluent<T, T> SearchFor(FilterDefinition<T> filter)
    {
        return Collection.Find(filter);
    }

    public IFindFluent<T, T> SearchFor(FilterDefinition<T> filter, FindOptions options)
    {
        return Collection.Find(filter, options);
    }

    public async IAsyncEnumerable<T> SearchForAsync(
        FilterDefinition<T> filter,
        [EnumeratorCancellation] CancellationToken cancellationToken = default
    )
    {
        var result = await Collection.FindAsync(filter, cancellationToken: cancellationToken);
        while (await result.MoveNextAsync(cancellationToken))
        {
            foreach (var item in result.Current)
            {
                yield return item;
            }
        }
    }

    public async IAsyncEnumerable<T> SearchForAsync(
        FilterDefinition<T> filter,
        FindOptions<T> options,
        [EnumeratorCancellation] CancellationToken cancellationToken = default
    )
    {
        var result = await Collection.FindAsync(filter, options, cancellationToken);
        while (await result.MoveNextAsync(cancellationToken))
        {
            foreach (var item in result.Current)
            {
                yield return item;
            }
        }
    }

    public async Task<T?> FindAsync(object id, CancellationToken cancellationToken = default)
    {
        var filter = MongodbExtensions.GetIdPropertyFilter<T>(id);
        return await (
            await Collection.FindAsync(filter, cancellationToken: cancellationToken)
        ).SingleOrDefaultAsync(cancellationToken: cancellationToken);
    }

    public async Task InsertAsync(T entity, CancellationToken cancellationToken = default)
    {
        await Collection.InsertOneAsync(session, entity, cancellationToken: cancellationToken);
    }

    public async Task InsertAsync(
        IReadOnlyCollection<T> source,
        CancellationToken cancellationToken = default
    )
    {
        if (source.Count > 0)
        {
            await Collection.InsertManyAsync(session, source, cancellationToken: cancellationToken);
        }
    }

    public async Task<DeleteResult> DeleteAsync(
        Expression<Func<T, bool>> filter,
        CancellationToken cancellationToken = default
    )
    {
        var result = await Collection.DeleteManyAsync(
            session,
            filter,
            cancellationToken: cancellationToken
        );
        return result;
    }

    public async Task<DeleteResult> DeleteAsync(
        FilterDefinition<T> filter,
        CancellationToken cancellationToken = default
    )
    {
        var result = await Collection.DeleteManyAsync(
            session,
            filter,
            cancellationToken: cancellationToken
        );
        return result;
    }

    public async Task<DeleteResult> DeleteAsync(
        object id,
        CancellationToken cancellationToken = default
    )
    {
        var filter = MongodbExtensions.GetIdPropertyFilter<T>(id);
        return await Collection.DeleteOneAsync(
            session,
            filter,
            cancellationToken: cancellationToken
        );
    }

    public async Task<UpdateResult> UpdateAsync(
        Expression<Func<T, bool>> predicate,
        UpdateDefinition<T> update,
        CancellationToken cancellationToken = default
    )
    {
        var result = await Collection.UpdateManyAsync(
            session,
            predicate,
            update,
            cancellationToken: cancellationToken
        );
        return result;
    }

    public async Task<ReplaceOneResult> UpdateAsync(
        T entity,
        CancellationToken cancellationToken = default
    )
    {
        var filter = MongodbExtensions.GetIdPropertyFilter(entity);
        var result = await Collection.ReplaceOneAsync(
            session,
            filter,
            entity,
            cancellationToken: cancellationToken
        );
        return result;
    }

    public async Task<BulkWriteResult<T>> UpdateAsync(
        IEnumerable<T> entities,
        CancellationToken cancellationToken = default
    )
    {
        var writes = entities
            .Select(x => new ReplaceOneModel<T>(MongodbExtensions.GetIdPropertyFilter(x), x))
            .ToList();

        return await Collection.BulkWriteAsync(
            session,
            writes,
            cancellationToken: cancellationToken
        );
    }

    public IMongoDatabase Database => access.Database;
}

⚠⚠ 以下内容为AI分析的结果，请根据实际情况进行判断。

这是一个基于 MongoDB 的工作单元模式（Unit of Work）的仓储实现类。让我详细解释一下这个代码的结构和功能：

类定义和构造函数

public sealed class RepositoryUnitOfWork<T>(MongodbAccess<T> access, IClientSessionHandle session)
    : IRepository<T>
    where T : IBaseEntity

这是一个密封的泛型类，使用了 C# 12 的主构造函数语法
泛型约束 T : IBaseEntity 确保操作的实体类型实现了基础实体接口
实现了 IRepository<T> 接口
接收两个参数：MongoDB访问对象和会话句柄（用于事务支持）

核心属性

public IMongoCollection<T> Collection => access.Collection;
public IMongoDatabase Database => access.Database;

提供对 MongoDB 集合和数据库的访问。

查询功能

1. LINQ 查询支持

public IQueryable<T> MongodbQueryable =>
    Database
        .GetCollection<T>(access.CollectionName)
        .AsQueryable(session, new AggregateOptions { AllowDiskUse = true });

提供 LINQ 查询能力
使用会话确保事务一致性
AllowDiskUse = true 允许大型聚合操作使用磁盘

2. 多种查询方法

SearchFor(Expression<Func<T, bool>>): LINQ 表达式查询
SearchFor(FilterDefinition<T>): MongoDB 原生过滤器查询
SearchForAsync: 异步枚举查询，支持流式处理大数据集

CRUD 操作

查找操作

public async Task<T?> FindAsync(object id, CancellationToken cancellationToken = default)

根据 ID 查找单个实体。

插入操作

InsertAsync(T entity): 插入单个实体
InsertAsync(IReadOnlyCollection<T> source): 批量插入实体

删除操作

支持三种删除方式：

基于 LINQ 表达式删除
基于 MongoDB 过滤器删除
基于 ID 删除

更新操作

提供多种更新方式：

UpdateAsync(predicate, update): 基于条件的部分更新
UpdateAsync(T entity): 替换整个实体
UpdateAsync(IEnumerable<T> entities): 批量更新实体

关键特性

1. 事务支持

所有写操作都使用 session 参数，确保在事务上下文中执行。

2. 异步流处理

public async IAsyncEnumerable<T> SearchForAsync(...)
{
    var result = await Collection.FindAsync(filter, cancellationToken: cancellationToken);
    while (await result.MoveNextAsync(cancellationToken))
    {
        foreach (var item in result.Current)
        {
            yield return item;
        }
    }
}

使用 yield return 实现流式数据处理，适合处理大量数据。

3. 取消令牌支持

所有异步方法都支持 CancellationToken，可以优雅地取消长时间运行的操作。

4. 批量操作优化

批量插入使用 InsertManyAsync
批量更新使用 BulkWriteAsync 和 ReplaceOneModel

设计模式

这个类实现了：

仓储模式：封装数据访问逻辑
工作单元模式：通过会话管理事务边界
异步编程模式：所有 I/O 操作都是异步的

这种设计使得数据访问层既灵活又高效，特别适合需要事务支持的企业级应用。

评论加载中...