Про паттерн

Часто приходится оперировать набором сущностей и моделей, для управления которыми создается также большое количество репозиториев. Паттерн Unit of Work помогает упростить работу с различными репозиториями и дает уверенность, что все репозитории будут использовать один и тот же DbContext. Но это всё про EntityFramework и значит про реляционные базы данных. А что же MongoDb?...

По-другому

Представленный вашему вниманию проект - всего лишь попытка упростить жизнь тем, кто использует MongoDb в повседневной работе, ну, или просто использует достаточно часто. Unit Of Work прекрасно работает с реляциями, но MongoDb - документо-ориентированная база данных, а это значит, что она работает по-другому. Возникла идея о том, как можно этот паттерн "натянуть" на подобную базу данных.

MongoDb.Driver

Обязательно надо сказать, что сборка использует другой nuget-пакет, который называется MongoDB.Driver. Просто его возможности немного расширены.

Возможности

Надо сказать честно, что не всё до конца получилось так, как задумывалось. И вс это потому, что с учем специфики работы самой MongoDb. Но есть полезные штуки, которые могут быть действительные полезны. Попробую перечислить то, что уже реализовано:

• Настройка подключения через appSettings.json
  • ConnectionString
  • набор параметров MongoClientSettings
• MongoDbVerboseLogging
• Тестирование подключение Transactions (ReplicaSet)
• Полный доступ к Collection (любые CRUD операции из MongoDB.Driver)
• Получение IClientSessionHandle по требованию
• Постраничная разбивка на страницы
• Получение в Repository сущности другие репозитории для других сущностей
•  

Три простых шага

Для начала скажу, что IUnitOfWork создавался в первую очередь для того, чтобы его можно было легко получить через вливание зависимостей, то есть для использования Dependency Injection. Всё что нужно сделать - это добавить секцию настроек в appSettings.json, зарегистрировать в контейнере (ServiceCollection) и начать использовать.

Шаг 1: appsettings.json

Пример настроек для подключения к localhost, без использования ssl и т.д. и т.п.

{
    "DatabaseSettings": {
        "ConnectionString": "mongodb://localhost:27017/?readPreference=primary&ssl=false&directConnection=true",
        "Credential": {
            "Login": "sa",
            "Password": "P@55w0rd"
        },
        "ApplicationName": "CalabongaDemo",
        "ReplicaSetName": "rs0",
        "DatabaseName": "MyDatabase",
        "Hosts": [ "localhost" ],
        "MongoDbPort": 27017,
        "VerboseLogging": false,
        "DirectConnection": true
    }
}

В данном пример указано ConnectionString, которая переопределит все настройки указанные ниже. Просто есть есть ConnectionString - остальные настройки игнорируются. В остальных случаях всё как обычно, и, надо сказать, это далеко неполный набор параметров, который используется для создания подключения. По мере необходимости набор может быть расширен.

Использование appSettings.json не обязательно. Подключить можно и с фиксированными настройками (hardcoded), задав конфигурацию прямо в коде.

Шаг 2: регистрация в DI-контейнере

Подключить можно двумя способами. Первый - это прочитав настройки из секции appSettings.json:

// read configuration section DatabaseSettings
services.AddUnitOfWork(configuration.GetSection(nameof(DatabaseSettings)));

Bторой способ не требует наличия appSettings.json, просто задайте параметры в коде:

services.AddUnitOfWork(config =>
{
    config.Credential = new CredentialSettings { Login = "sa", Password = "password" };
    config.DatabaseName = "MyDatabase";
    config.Hosts = new[] { "Localhost" };
    config.MongoDbPort = 27017;
    config.VerboseLogging = false;
});

Шаг 3: использование

Можно внедрить зависимость в PageModel, например, так:

namespace WebApplicationWithMongo.Pages
{
    public class IndexModel : PageModel
    {
        private readonly IUnitOfWork _unitOfWork;
        private readonly ILogger<IndexModel> _logger;

        public IndexModel(IUnitOfWork unitOfWork, ILogger<IndexModel> logger)
        {
            _unitOfWork = unitOfWork;
            _logger = logger;
        }

        public IPagedList<Order>? Data { get; set; }
    }
}

Теперь в методе можно использовать _unitOfWork, например, для получения коллекции объектов с разбиение на страницы (paged data):

public async Task<IActionResult> OnGetAsync(int pageIndex = 0, int pageSize = 10)
{
    var repository = _unitOfWork.GetRepository<Order, int>();
    Data = await repository.GetPagedAsync(pageIndex, pageSize, FilterDefinition<Order>.Empty, HttpContext.RequestAborted);
    return Page();
}

В данном примере я использую объекты Order:

[BsonIgnoreExtraElements]
public class Order : DocumentBase<int>
{
    [BsonElement("number")]
    [BsonRepresentation(BsonType.Int32)]
    public int Number { get; set; }

    [BsonElement("title")]
    [BsonRepresentation(BsonType.String)]
    public string Title { get; set; } = default!;

    [BsonElement("description")]
    [BsonRepresentation(BsonType.String)]
    public string? Description { get; set; }

    [BsonElement("items")]
    public ICollection<OrderItem>? Items { get; set; }
}

и OrderItem:

public class OrderItem : DocumentBase<int>
{
    [BsonElement("name")]
    public string Name { get; set; } = default!;

    [BsonElement("quantity")]
    [BsonRepresentation(BsonType.Int32)]
    public int Quantity { get; set; }

    [BsonElement("price")]
    [BsonRepresentation(BsonType.Double)]
    public double Price { get; set; }
}

А вот что получилось в итоге:

Транзакции в MongoDb

Для примера хочу показать как использовать транзакции в MongoDb в моей библиотеке. Для начала нужно определить при каких условиях возможен вызов методов CRUD в MongoDb.

• База данных MongoDb должна быть настроена соответствующим образом. То есть поддерживать транзакции. Как настроить транзакции в MongoDb можно посмотреть в статье Configuring ReplicaSet. Если MongoDb запущен в режиме Statenalone - транзакции работать не будут. Для проверки работоспособности транзакций можно воспользоваться методом EnsureReplicationSetReady.
• Вызов методов, которые должы быть выполнены в транзакции следует через специальные методы UseTransactionAsync или другие его перегрузки.

Пример, как вызывать методы с транзакциями:

// предположим, что методы вызываются в контроллере API

await unitOfWork.UseTransactionAsync<OrderBase, int>(ProcessDataInTransactionAsync1, HttpContext.RequestAborted, session);

await unitOfWork.UseTransactionAsync(ProcessDataInTransactionAsync2, repository, HttpContext.RequestAborted, session);

await unitOfWork.UseTransactionAsync(ProcessDataInTransactionAsync3, repository, new TransactionContext(new TransactionOptions(), session, HttpContext.RequestAborted));

await unitOfWork.UseTransactionAsync<OrderBase, int>(ProcessDataInTransactionAsync4, new TransactionContext(new TransactionOptions(), session, HttpContext.RequestAborted));

await unitOfWork.UseTransactionAsync<OrderBase, int>(ProcessDataInTransactionAsync5, TransactionContext.Default);

А теперь как примерно должны выглядеть сами методы, которые вызываются:

async Task ProcessDataInTransactionAsync1(IRepository<OrderBase, int> repositoryInTransaction, IClientSessionHandle session, CancellationToken cancellationToken)
{
    await repository.Collection.DeleteManyAsync(session, FilterDefinition<OrderBase>.Empty, null, cancellationToken);

    var internalOrder1 = DocumentHelper.GetInternal(99);
    await repositoryInTransaction.Collection.InsertOneAsync(session, internalOrder1, null, cancellationToken);
    logger!.LogInformation("InsertOne: {item1}", internalOrder1);

    var internalOrder2 = DocumentHelper.GetInternal(100);
    await repositoryInTransaction.Collection.InsertOneAsync(session, internalOrder2, null, cancellationToken);
    logger!.LogInformation("InsertOne: {item2}", internalOrder2);

    var filter = Builders<OrderBase>.Filter.Eq(x => x.Id, 99);
    var updateDefinition = Builders<OrderBase>.Update.Set(x => x.Description, "Updated description");
    var result = await repositoryInTransaction.Collection
        .UpdateOneAsync(session, filter, updateDefinition, new UpdateOptions { IsUpsert = false }, cancellationToken);

    if (result.IsModifiedCountAvailable)
    {
        logger!.LogInformation("Update {}", result.ModifiedCount);
    }

    throw new ApplicationException("EXCEPTION! BANG!");
}

async Task ProcessDataInTransactionAsync2(IRepository<OrderBase, int> repositoryInTransaction, IClientSessionHandle session, CancellationToken cancellationToken)
{
    await repositoryInTransaction.Collection.DeleteManyAsync(session, FilterDefinition<OrderBase>.Empty, null, cancellationToken);

    var internalOrder1 = DocumentHelper.GetInternal(99);
    await repositoryInTransaction.Collection.InsertOneAsync(session, internalOrder1, null, cancellationToken);
    logger!.LogInformation("InsertOne: {item1}", internalOrder1);

    var internalOrder2 = DocumentHelper.GetInternal(100);
    await repositoryInTransaction.Collection.InsertOneAsync(session, internalOrder2, null, cancellationToken);
    logger!.LogInformation("InsertOne: {item2}", internalOrder2);

    var filter = Builders<OrderBase>.Filter.Eq(x => x.Id, 99);
    var updateDefinition = Builders<OrderBase>.Update.Set(x => x.Description, "Updated description");
    var result = await repositoryInTransaction.Collection.UpdateOneAsync(session, filter, updateDefinition, new UpdateOptions { IsUpsert = false }, cancellationToken);

    if (result.IsModifiedCountAvailable)
    {
        logger!.LogInformation("Update {}", result.ModifiedCount);
    }

    throw new ApplicationException("EXCEPTION! BANG!");
}

async Task ProcessDataInTransactionAsync3(IRepository<OrderBase, int> repositoryInTransaction, TransactionContext transactionContext)
{
    await repositoryInTransaction.Collection.DeleteManyAsync(transactionContext.Session, FilterDefinition<OrderBase>.Empty, null, transactionContext.CancellationToken);

    var internalOrder1 = DocumentHelper.GetInternal(99);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder1, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item1}", internalOrder1);

    var internalOrder2 = DocumentHelper.GetInternal(100);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder2, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item2}", internalOrder2);

    var filter = Builders<OrderBase>.Filter.Eq(x => x.Id, 99);
    var updateDefinition = Builders<OrderBase>.Update.Set(x => x.Description, "Updated description");
    var result = await repositoryInTransaction.Collection.UpdateOneAsync(transactionContext.Session, filter, updateDefinition, new UpdateOptions { IsUpsert = false }, transactionContext.CancellationToken);

    if (result.IsModifiedCountAvailable)
    {
        transactionContext.Logger.LogInformation("Update {}", result.ModifiedCount);
    }

    throw new ApplicationException("EXCEPTION! BANG!");
}

async Task ProcessDataInTransactionAsync4(IRepository<OrderBase, int> repositoryInTransaction, TransactionContext transactionContext)
{
    await repositoryInTransaction.Collection.DeleteManyAsync(transactionContext.Session, FilterDefinition<OrderBase>.Empty, null, transactionContext.CancellationToken);

    var internalOrder1 = DocumentHelper.GetInternal(99);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder1, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item1}", internalOrder1);

    var internalOrder2 = DocumentHelper.GetInternal(100);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder2, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item2}", internalOrder2);

    var filter = Builders<OrderBase>.Filter.Eq(x => x.Id, 99);
    var updateDefinition = Builders<OrderBase>.Update.Set(x => x.Description, "Updated description");
    var updateResult = await repositoryInTransaction.Collection.UpdateOneAsync(transactionContext.Session, filter, updateDefinition, new UpdateOptions { IsUpsert = false }, transactionContext.CancellationToken);

    if (updateResult.IsModifiedCountAvailable)
    {
        transactionContext.Logger.LogInformation("Update {}", updateResult.ModifiedCount);
    }

    throw new ApplicationException("EXCEPTION! BANG!");
}

async Task ProcessDataInTransactionAsync5(IRepository<OrderBase, int> repositoryInTransaction, TransactionContext transactionContext)
{
    await repositoryInTransaction.Collection.DeleteManyAsync(transactionContext.Session, FilterDefinition<OrderBase>.Empty, null, transactionContext.CancellationToken);

    var internalOrder1 = DocumentHelper.GetInternal(99);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder1, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item1}", internalOrder1);

    var internalOrder2 = DocumentHelper.GetInternal(100);
    await repositoryInTransaction.Collection.InsertOneAsync(transactionContext.Session, internalOrder2, null, transactionContext.CancellationToken);
    transactionContext.Logger.LogInformation("InsertOne: {item2}", internalOrder2);

    var filter = Builders<OrderBase>.Filter.Eq(x => x.Id, 99);
    var updateDefinition = Builders<OrderBase>.Update.Set(x => x.Description, "Updated description");
    var updateResult = await repositoryInTransaction.Collection.UpdateOneAsync(transactionContext.Session, filter, updateDefinition, new UpdateOptions { IsUpsert = false }, transactionContext.CancellationToken);

    if (updateResult.IsModifiedCountAvailable)
    {
        transactionContext.Logger.LogInformation("Update {}", updateResult.ModifiedCount);
    }

    throw new ApplicationException("EXCEPTION! BANG!");
}

Настоятельно рекомендую ознакомиться с тем как используются методы вызова CRUD операций внутри каждого и методов. Обратите внимание на то, что обязательно использование session для вызова операций на операции Insert/Update/Delete. В противном случае, при ошибке или исключении отмены операции не произойдет.

Видео

Бонус

Для создания MongoDb в Docker я использовал docker-compose файл "docker-compose-mongo.yml". Этот yml создает Standalone сервер MongoDb:

version: '3.8'

services:

  mongo:
    image: mongo
    environment:
      MONGO_INITDB_ROOT_USERNAME: mongo
      MONGO_INITDB_ROOT_PASSWORD: mongo
    volumes:
      - mongo_data:/data/db
    ports:
      - "27017:27017"

volumes:
  mongo_data: null

networks:
  default:
    external: false
    name: local

А для создания MongoDb в Docker c ReplicaSet я использовать docker-compose файл "". Здесь уже можно запускать транзакции на операции:

version: '3.8'

services:
  mongo1:
    container_name: mongo1
    image: mongo
    networks:
      - mongo-network
    ports:
      - 27017:27017
    depends_on:
      - mongo2
      - mongo3
    links:
      - mongo2
      - mongo3
    restart: always
    entrypoint:
      [
        "/usr/bin/mongod",
        "--bind_ip_all",
        "--replSet",
        "rs0"
      ]

  mongo2:
    container_name: mongo2
    image: mongo
    networks:
      - mongo-network
    ports:
      - 27018:27017
    restart: always
    entrypoint:
      [
        "/usr/bin/mongod",
        "--bind_ip_all",
        "--replSet",
        "rs0"
      ]
  mongo3:
    container_name: mongo3
    image: mongo
    networks:
      - mongo-network
    ports:
      - 27019:27017
    restart: always
    entrypoint:
      [
        "/usr/bin/mongod",
        "--bind_ip_all",
        "--replSet",
        "rs0"
      ]

networks:
  mongo-network:
    driver: bridge

Настройка ReplicaSet

rs.initiate(
   {
      _id: "rs0",
      version: 1,
      members: [
         { _id: 0, host : "mongo1:27017", priority: 4 },
         { _id: 1, host : "mongo2:27017", priority: 2 },
         { _id: 2, host : "mongo3:27017", priority: 1 }
      ]
   }
)

Создание пользователя для ReplicaSet

use admin

db.createUser({ user: "admin", pwd: "admin", roles: [ { role: "root", db: "admin" } ] })

Строка для подключения

mongodb://mongo1:27017,mongo2:27017,mongo3:27017/?replicaSet=rs0&ssl=false

Ссылки

Nuget пакет Calabonga.UnitOfWork.MongoDb

github Calabonga.UnitOfWork.MongoDb

Configure ReplicaSet docker-compose-mongo-replica-set

Решение ошибки Reconfigure a Replica Set with Unavailable Members ("not primary")