Блог

У меня есть следующий вспомогательный класс, который я использую для выполнения одной задачи асинхронной инициализации ( CreateKafkaTopic ) перед созданием тестового сервера через WebApplicationFactory.

Вспомогательный класс для создания темы Kafka из конфигурации с помощью метода CreateKafkaTopic

 using System;
using System.Threading.Tasks;

using Autofac;
using Autofac.Builder;
using Autofac.Extensions.DependencyInjection;
using Extensions.Hosting.AsyncInitialization;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;

using WebApp.Data.Serializers.AutofacModule;
using WebApp.Kafka.AutofacModule;
using WebApp.Kafka.Admin;
using WebApp.Kafka.Config;
using WebApp.Mqtt.Config;
using WebApp.S3.Config;


namespace WebApp.Testing.Fixtures
{
    /// <summary>
    /// Core helper testing utilities
    /// </summary>
    public class CoreTestUtils
    {
        /// <summary>
        /// Create a configuration object from appsettings.json file
        /// </summary>
        /// <param name="basePath">Base path for json file</param>
        /// <param name="fileName">Name of the json file</param>
        public static IConfigurationRoot GetIConfigurationRoot(string basePath, string fileName = "appsettings")
        {
            if (basePath == null)
                throw new ArgumentNullException(nameof(basePath));

            return new ConfigurationBuilder()
                .SetBasePath(basePath)
                .AddJsonFile($"{fileName}.json", optional: true)
                .AddEnvironmentVariables()
                .Build();
        }

        /// <summary>
        /// Configure services for logging, signalR and config options
        /// </summary>
        /// <param name="configBasePath">Base path for json file</param>
        /// <param name="fileName">Filename for json file, defaults to 'appsettings'</param>
        /// <returns>Populated service collection</returns>
        public static IServiceCollection ConfigureServices(string configBasePath, string fileName = "appsettings")
        {
            if (configBasePath == null)
                throw new ArgumentNullException(nameof(configBasePath));

            var services = new ServiceCollection();

            // Add configuration from appsettings.json
            IConfigurationRoot _configuration = CoreTestUtils.GetIConfigurationRoot(configBasePath, fileName);
            services.Configure<KafkaConfig>(options =>
                _configuration.GetSection(KafkaConfig.SectionName).Bind(options));
            services.Configure<S3Config>(options =>
                _configuration.GetSection(S3Config.SectionName).Bind(options));
            services.Configure<MqttConfig>(options =>
                _configuration.GetSection(MqttConfig.SectionName).Bind(options));

            // Populate service provider with usual services
            services.AddLogging();
            services.AddSignalR();
            services.AddOptions();
            services.AddSingleton<IKafkaAdminFactory, KafkaAdminFactory>();
            services.AddAsyncInitializer<KafkaAdminService>();

            return services;
        }

        /// <summary>
        /// Create topic on Kafka cluster based on settings in config file
        /// </summary>
        /// <param name="configBasePath">Base path where config file exists</param>
        /// <param name="fileName">The config file that exists in the base path</param>
        /// <exception cref="ArgumentNullException">Thrown if KafkaAdmin service failed to be resolved</exception>
        public static async Task CreateKafkaTopic(string configBasePath, string fileName = "appsettings")
        {
            Console.WriteLine($"Trying to create a Kafka Topic with basePath:{configBasePath} and file:{fileName}");

            // environment variables are set at this point

            var services = ConfigureServices(configBasePath, fileName);

            using (var container = CoreTestUtils.ConfigureContainer(services, new Autofac.Module[] {
                new SerializerModule(),
                new KafkaModule() }))
            {
                using (var scope = container.BeginLifetimeScope())
                {
                    Console.WriteLine("About to create the factory");
                    var factory = scope.Resolve<IKafkaAdminFactory>();
                    Console.WriteLine("Have created the factory");
                    var client = factory.CreateAdminClient();
                    Console.Write($"client is null? => {client == null}");
                    await Task.CompletedTask;
                }
            }
        }


        /// <summary>
        /// Build Autofac container with the modules and services provided
        /// </summary>
        /// <param name="services">Services</param>
        /// <param name="modules">List of Autofac modules</param>
        /// <returns>Populated service collection</returns>
        public static IContainer ConfigureContainer(IServiceCollection services, Autofac.Module[] modules)
        {
            if (services == null)
                throw new ArgumentNullException(nameof(services));

            if (modules == null)
                throw new ArgumentNullException(nameof(modules));

            var containerBuilder = new ContainerBuilder();

            foreach (var module in modules)
            {
                containerBuilder.RegisterModule(module);
            }

            containerBuilder.Populate(services);

            return containerBuilder.Build(ContainerBuildOptions.IgnoreStartableComponents);
        }
    }
}
 

Задача инициализации создает kafka topic ( CreateKafkaTopic ) из конфигурации, используя службу KafkaAdminService (см. Список ниже). Служба имеет зависимость от заводского класса ( IKafkaAdminFactory ) для создания объединенного IAdminClient кафки на основе зависимости IOptions.

Я изо всех сил пытался инициализировать зависимость IOptions с помощью моего вспомогательного класса. Похоже, что зависимость IOptions не заполняет мой класс POCO из файла, вызывая исключение, что требуемый раздел отсутствует. У кого-нибудь есть какие-либо идеи, почему конфигурация POCO не инициализируется из файла appsettings с помощью CreateKafkaTopic моего вспомогательного класса?

KafkaAdminFactory

  public class KafkaAdminFactory : IKafkaAdminFactory
    {
        private KafkaConfig _Config { get; }
        public KafkaAdminFactory(IOptions<KafkaConfig> options)
        {
            _Config = options.Value ?? throw new ArgumentNullException(nameof(options));
        }

        public IAdminClient CreateAdminClient()
        {
            var adminClientBuilder = new AdminClientBuilder(
                new AdminClientConfig()
                {
                    BootstrapServers = _Config.Consumer.BootstrapServers
                }
            );

            return adminClientBuilder.Build();
        }
 

KafkaAdminService

 using System;
using System.Threading.Tasks;

using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Extensions.Hosting.AsyncInitialization;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

using WebApp.Kafka.Config;


namespace WebApp.Kafka.Admin
{
    /// <summary> Use a factory method to create the client, increased testability </summary>
    public delegate IAdminClient KafkaAdminFactoryD(KafkaConfig config);

    /// <summary>Service to make a request from Kafka to create a topic</summary>
    public class KafkaAdminService : IAsyncInitializer
    {
        private IKafkaAdminFactory _Factory { get; set; }
        private ILogger<KafkaAdminService> _Logger { get; set; }
        private KafkaConfig _Config { get; set; }


        /// <summary>
        /// Retrieve KafkaConfig from appsettings
        /// </summary>
        /// <param name="config">Config POCO from appsettings file</param>
        /// <param name="clientFactory"><see cref="KafkaAdminFactory"/></param>
        /// <param name="logger">Logger instance</param>
        public KafkaAdminService(
            IOptions<KafkaConfig> config,
            IKafkaAdminFactory clientFactory,
            ILogger<KafkaAdminService> logger)
        {
            if (clientFactory == null)
                throw new ArgumentNullException(nameof(clientFactory));

            if (config == null)
                throw new ArgumentNullException(nameof(config));

            _Config = config.Value ?? throw new ArgumentNullException(nameof(config));
            _Factory = clientFactory ?? throw new ArgumentNullException(nameof(clientFactory));
            _Logger = logger ?? throw new ArgumentNullException(nameof(logger));
        }


        /// <summary>
        /// Create a Kafka topic if it does not already exist
        /// </summary>
        /// <exception name="CreateTopicsException">
        /// Thrown for exceptions encountered except duplicate topic
        /// </exception>
        public async Task InitializeAsync()
        {
            using (var client = _Factory.CreateAdminClient())
            {
                try
                {
                    _Logger.LogInformation("Admin service trying to create Kafka Topic...");
                    _Logger.LogInformation($"Topic::{_Config.Topic.Name}, ReplicationCount::{_Config.Topic.ReplicationCount}, PartitionCount::{_Config.Topic.PartitionCount}");
                    _Logger.LogInformation($"Bootstrap Servers::{_Config.Consumer.BootstrapServers}");

                    await client.CreateTopicsAsync(new TopicSpecification[] {
                        new TopicSpecification {
                            Name = _Config.Topic.Name,
                            NumPartitions = _Config.Topic.PartitionCount,
                            ReplicationFactor = _Config.Topic.ReplicationCount
                        }
                    }, null);

                    _Logger.LogInformation($"Admin service successfully created topic {_Config.Topic.Name}");
                }
                catch (CreateTopicsException e)
                {
                    if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists)
                    {
                        _Logger.LogInformation($"An error occured creating topic {_Config.Topic.Name}: {e.Results[0].Error.Reason}");
                        throw e;
                    }
                    else
                    {
                        _Logger.LogInformation($"Topic {_Config.Topic.Name} already exists");
                    }
                }
            }

            _Logger.LogInformation("Kafka Consumer thread started");

            await Task.CompletedTask;
        }
    }
}
 

Решаемая проблема. Я указал неправильный путь к своему конфигурационному файлу и также считывал файл настроек как необязательный. Переработал мой код следующим образом для использования, чтобы помочь выполнить задачу асинхронной инициализации при использовании WebApplicationFactory :

 using System;
using System.IO;
using System.Threading.Tasks;

using Autofac;
using Autofac.Builder;
using Autofac.Extensions.DependencyInjection;
using Extensions.Hosting.AsyncInitialization;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;

using WebApp.Kafka.AutofacModule;
using WebApp.Kafka.Admin;
using WebApp.Kafka.Config;
using WebApp.Mqtt.Config;
using WebApp.S3.Config;


namespace WebApp.Testing.Fixtures
{
    /// <summary>
    /// Core helper testing utilities
    /// </summary>
    public class CoreTestUtils
    {
        /// <summary>
        /// Create a configuration object from appsettings.json file and environmental variables
        /// </summary>
        /// <param name="basePath">Base path for json file</param>
        /// <param name="fileName">Name of the json file</param>
        /// <param name="optional">True when json config file is mandatory</param>
        public static IConfigurationRoot GetIConfigurationRoot(string basePath, string fileName = "appsettings.json", bool optional = true)
        {
            if (basePath == null)
                throw new ArgumentNullException(nameof(basePath));

            string fullPath = Path.Combine(basePath, $"{fileName}");
            return new ConfigurationBuilder()
                .AddJsonFile(fullPath, optional: optional)
                .AddEnvironmentVariables()
                .Build();
        }

        /// <summary>
        /// Configure services for logging, signalR and config options
        /// </summary>
        /// <param name="configBasePath">Base path for json file</param>
        /// <param name="fileName">Filename for json file, defaults to 'appsettings'</param>
        /// <param name="optional">True when json config file is mandatory</param>
        /// <returns>Populated service collection</returns>
        public static IServiceCollection ConfigureServices(string configBasePath, string fileName = "appsettings.json", bool optional = true)
        {
            if (configBasePath == null)
                throw new ArgumentNullException(nameof(configBasePath));

            var services = new ServiceCollection();

            // Add configuration from appsettings.json
            IConfigurationRoot _configuration = CoreTestUtils.GetIConfigurationRoot(configBasePath, fileName, optional);
            services.Configure<KafkaConfig>(options =>
                _configuration.GetSection(KafkaConfig.SectionName).Bind(options));
            services.Configure<S3Config>(options =>
                _configuration.GetSection(S3Config.SectionName).Bind(options));
            services.Configure<MqttConfig>(options =>
                _configuration.GetSection(MqttConfig.SectionName).Bind(options));

            // Populate service provider with usual services
            services.AddLogging();
            services.AddSignalR();
            services.AddOptions();
            services.AddSingleton<IKafkaAdminFactory, KafkaAdminFactory>();
            services.AddAsyncInitializer<KafkaAdminService>();

            return services;
        }

        /// <summary>
        /// Create topic on Kafka cluster based on settings in config file
        /// </summary>
        /// <param name="configBasePath">Base path where config file exists</param>
        /// <param name="fileName">The config file that exists in the base path</param>
        /// <param name="optional">True when json config file is mandatory</param>
        /// <exception cref="ArgumentNullException">Thrown if KafkaAdmin service failed to be resolved</exception>
        public static async Task CreateKafkaTopic(string configBasePath, string fileName = "appsettings.json", bool optional = true)
        {
            var services = ConfigureServices(configBasePath, fileName, optional);

            using (var container = CoreTestUtils.ConfigureContainer(services, new Autofac.Module[] { new KafkaModule() }))
            {
                using (var scope = container.BeginLifetimeScope())
                {
                    var kafkaSvc = scope.Resolve<IAsyncInitializer>() ?? throw new ArgumentNullException("Failed to resolve KafkaAdminService");
                    await kafkaSvc.InitializeAsync();
                }
            }
        }

        /// <summary>
        /// Build Autofac container with the modules and services provided
        /// </summary>
        /// <param name="services">Services</param>
        /// <param name="modules">List of Autofac modules</param>
        /// <returns>Populated service collection</returns>
        public static IContainer ConfigureContainer(IServiceCollection services, Autofac.Module[] modules)
        {
            if (services == null)
                throw new ArgumentNullException(nameof(services));

            if (modules == null)
                throw new ArgumentNullException(nameof(modules));

            var containerBuilder = new ContainerBuilder();

            foreach (var module in modules)
            {
                containerBuilder.RegisterModule(module);
            }

            containerBuilder.Populate(services);

            return containerBuilder.Build(ContainerBuildOptions.IgnoreStartableComponents);
        }
    }
}