ECommon.AspectCore 1.0.1

In startUp.cs

public IServiceProvider ConfigureServices(IServiceCollection services)
{
  services.AddMvc();
  var container = services.ToServiceContainer();
  InitializeENodeFramework(container);
  return (ObjectContainer.Current as AspectCoreObjectContainer)?.ServiceResolver;
}

static void InitializeENodeFramework(IServiceContainer serviceContainer)
{
  var assemblies = new[]
  {
    Assembly.Load("ENodeDemo.Domain"),
    Assembly.Load("ENodeDemo.Commands"),
    Assembly.Load("ENodeDemo.CommandHandlers"),
    Assembly.GetExecutingAssembly()
  };
  ECommon.Configurations.Configuration
    .Create()
    .UseAspectCore(serviceContainer)
    .RegisterCommonComponents()
    .UseLog4Net()
    .UseJsonNet()
    .RegisterUnhandledExceptionHandler()
    .CreateENode()
    .RegisterENodeComponents()
    .RegisterBusinessComponents(assemblies)
    .UseEQueue()
    .InitializeMySqlEventStore("")
    .InitializeMySqlLockService("")
    .InitializeMySqlPublishedVersionStore("")
    .UseMySqlEventStore()
    .UseMySqlLockService()
    .UseMySqlPublishedVersionStore()
    .BuildContainer()
    .InitializeBusinessAssemblies(assemblies)
    .StartEQueue()
    .Start();
}

this is ENodeExtensions

public static class ENodeExtensions
{
  private static NameServerController _nameServerController;
  private static BrokerController _broker;
  private static CommandService _commandService;
  private static CommandConsumer _commandConsumer;
  private static DomainEventPublisher _eventPublisher;
  private static DomainEventConsumer _eventConsumer;

  public static ENodeConfiguration BuildContainer(this ENodeConfiguration enodeConfiguration)
  {
    enodeConfiguration.GetCommonConfiguration().BuildContainer();
    return enodeConfiguration;
  }
  
  public static ENodeConfiguration UseEQueue(this ENodeConfiguration enodeConfiguration)
  {
    var assemblies = new[] { Assembly.GetExecutingAssembly() };
    enodeConfiguration.RegisterTopicProviders(assemblies);

    var configuration = enodeConfiguration.GetCommonConfiguration();
    configuration.RegisterEQueueComponents();

    _commandService = new CommandService();
    _eventPublisher = new DomainEventPublisher();

    configuration.SetDefault<ICommandService, CommandService>(_commandService);
    configuration.SetDefault<IMessagePublisher<DomainEventStreamMessage>, DomainEventPublisher>(_eventPublisher);
    
    return enodeConfiguration;
  }
  
  public static ENodeConfiguration StartEQueue(this ENodeConfiguration enodeConfiguration)
  {
    var configuration = enodeConfiguration.GetCommonConfiguration();
   
    _commandService.Initialize(new CommandResultProcessor().Initialize(new IPEndPoint(SocketUtils.GetLocalIPV4(), 9000)));
    _eventPublisher.Initialize();
    _commandConsumer = new CommandConsumer().Initialize().Subscribe("NoteCommandTopic");
    _eventConsumer = new DomainEventConsumer().Initialize().Subscribe("NoteEventTopic");

    var brokerStorePath = @"c:\equeue-store";
    if (Directory.Exists(brokerStorePath))
    {
      Directory.Delete(brokerStorePath, true);
    }
    
    _nameServerController = new NameServerController();
    _broker = BrokerController.Create();
    _nameServerController.Start();
    _broker.Start();
    _eventConsumer.Start();
    _commandConsumer.Start();
    _eventPublisher.Start();
    _commandService.Start();

    WaitAllConsumerLoadBalanceComplete();
    return enodeConfiguration;
  }
  
  public static ENodeConfiguration ShutdownEQueue(this ENodeConfiguration enodeConfiguration)
  {
    _commandService.Shutdown();
    _eventPublisher.Shutdown();
    _commandConsumer.Shutdown();
    _eventConsumer.Shutdown();
    _broker.Shutdown();
    _nameServerController.Shutdown();
    return enodeConfiguration;
  }
  
  private static void WaitAllConsumerLoadBalanceComplete()
  {
    var logger = ObjectContainer.Resolve<ILoggerFactory>().Create(typeof(ENodeExtensions).Name);
    var scheduleService = ObjectContainer.Resolve<IScheduleService>();
    var waitHandle = new ManualResetEvent(false);
    logger.Info("Waiting for all consumer load balance complete, please wait for a moment...");
    scheduleService.StartTask("WaitAllConsumerLoadBalanceComplete", () =>
    {
      var eventConsumerAllocatedQueues = _eventConsumer.Consumer.GetCurrentQueues();
      var commandConsumerAllocatedQueues = _commandConsumer.Consumer.GetCurrentQueues();
      if (eventConsumerAllocatedQueues.Count() == 4 && commandConsumerAllocatedQueues.Count() == 4)
      {
        waitHandle.Set();
       }
     }, 1000, 1000);
     
     waitHandle.WaitOne();
     scheduleService.StopTask("WaitAllConsumerLoadBalanceComplete");
     logger.Info("All consumer load balance completed.");
   }
}