JiShe.CollectBus/src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs

using System.Linq;
using Microsoft.Extensions.DependencyInjection;
using Volo.Abp.AutoMapper;
using Volo.Abp.Modularity;
using Volo.Abp.Application;
using Volo.Abp.BackgroundWorkers;
using System.Threading.Tasks;
using Volo.Abp;
using System.Reflection;
using JiShe.CollectBus.FreeSql;
using System;
using JiShe.CollectBus.Common.Extensions;
using Volo.Abp.AspNetCore.Mvc.AntiForgery;
using JiShe.CollectBus.FreeRedisProvider;
using JiShe.CollectBus.Workers;
using Volo.Abp.BackgroundWorkers.Hangfire;
using JiShe.CollectBus.MongoDB;
using JiShe.CollectBus.ScheduledMeterReading;
using AutoMapper.Configuration.Annotations;
using JiShe.CollectBus.Common.Attributes;
using JiShe.CollectBus.IoTDBProvider;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.Protocol.Contracts;
using System.Collections.Generic;
using Thrift;
using Microsoft.Extensions.Configuration;
using Volo.Abp.EventBus.Kafka;
using Volo.Abp.Kafka;
using Volo.Abp.EventBus;
using Confluent.Kafka;
using JiShe.CollectBus.Kafka.AdminClient;
using JiShe.CollectBus.Common.Consts;

namespace JiShe.CollectBus;

[DependsOn(
    typeof(CollectBusDomainModule),
    typeof(CollectBusApplicationContractsModule),
    typeof(AbpDddApplicationModule),
    typeof(AbpAutoMapperModule),
    typeof(AbpBackgroundWorkersHangfireModule),
    typeof(CollectBusFreeRedisModule),
    typeof(CollectBusFreeSqlModule),
    typeof(AbpEventBusModule), 
    typeof(AbpKafkaModule),
    typeof(CollectBusIoTDBModule)
    )]
public class CollectBusApplicationModule : AbpModule
{
    public override void ConfigureServices(ServiceConfigurationContext context)
    {
        var configuration = context.Services.GetConfiguration();

        context.Services.AddAutoMapperObjectMapper<CollectBusApplicationModule>();
        Configure<AbpAutoMapperOptions>(options =>
        {
            options.AddMaps<CollectBusApplicationModule>(validate: true);
        });

        Configure<AbpKafkaOptions>(configuration.GetSection("Kafka"));
        Configure<AbpKafkaEventBusOptions>(configuration.GetSection("Kafka:EventBus"));

        Configure<AbpKafkaOptions>(options =>
        {
            options.ConfigureConsumer = config =>
            {
                config.GroupId = configuration.GetValue<string>("Kafka:Consumer:GroupId");
                config.EnableAutoCommit = configuration.GetValue<bool>("Kafka:Consumer:EnableAutoCommit");
            };
        });

        Configure<AbpKafkaOptions>(options =>
        {
            options.ConfigureProducer = config =>
            {
                config.MessageTimeoutMs = configuration.GetValue<int>("Kafka:Producer:MessageTimeoutMs"); 
                config.Acks = (Acks)configuration.GetValue<int>("Kafka:Producer:Acks");
            };
        });

        Configure<AbpKafkaOptions>(options =>
        {
            options.ConfigureTopic = specification =>
            {
                specification.ReplicationFactor = configuration.GetValue<short>("Kafka:Topic:ReplicationFactor"); 
                specification.NumPartitions = configuration.GetValue<int>("Kafka:Topic:NumPartitions"); 
            };
        });
    }

    public override void OnApplicationInitialization(
        ApplicationInitializationContext context)
    {
        context.ServiceProvider.GetRequiredService<CustomKafkaDistributedEventBus>().Initialize();

        var assembly = Assembly.GetExecutingAssembly();
        var types = assembly.GetTypes().Where(t => typeof(ICollectWorker).IsAssignableFrom(t) && !t.IsInterface).ToList();
        foreach (var type in types)
        {
            context.AddBackgroundWorkerAsync(type);
        }

        //默认初始化表计信息
        var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
        dbContext.InitAmmeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult(); 
        dbContext.InitWatermeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult();

        //初始化主题信息
        var kafkaAdminClient = context.ServiceProvider.GetRequiredService<IAdminClientService>();
        var configuration = context.ServiceProvider.GetRequiredService<IConfiguration>();

        List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
        topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());

        foreach (var item in topics)
        {
            kafkaAdminClient.CreateTopicAsync(item, configuration.GetValue<int>(CommonConst.NumPartitions), configuration.GetValue<short>(CommonConst.KafkaReplicationFactor));
        }
    }

}