feat(core): gate concurrent datasource init by config

packages/core/src/common/dataSourceManager.ts

@@ -12,6 +12,7 @@ import { loadModule } from '../util';
 import { ModuleLoadType, DataSourceManagerConfigOption } from '../interface';
 import { Inject } from '../decorator';
 import { MidwayEnvironmentService } from '../service/environmentService';
+import { MidwayConfigService } from '../service/configService';
 import { MidwayPriorityManager } from './priorityManager';
 
 const debug = debuglog('midway:debug');
@@ -32,6 +33,9 @@ export abstract class DataSourceManager<
   @Inject()
   protected environmentService: MidwayEnvironmentService;
 
+  @Inject()
+  protected configService: MidwayConfigService;
+
   @Inject()
   protected priorityManager: MidwayPriorityManager;
 
@@ -57,11 +61,16 @@ export abstract class DataSourceManager<
         entitiesConfigKey: 'entities',
       };
     }
+    const initOptions = baseDirOrOptions;
+    const concurrent =
+      this.configService?.getConfiguration(
+        'core.concurrentDataSourceInitialization'
+      ) ?? false;
 
-    for (const dataSourceName in dataSourceConfig.dataSource) {
+    const processDataSource = async (dataSourceName: string) => {
       const dataSourceOptions = dataSourceConfig.dataSource[dataSourceName];
       const userEntities = dataSourceOptions[
-        baseDirOrOptions.entitiesConfigKey
+        initOptions.entitiesConfigKey
       ] as any[];
       if (userEntities) {
         const entities = new Set();
@@ -71,7 +80,7 @@ export abstract class DataSourceManager<
             // string will be glob file
             const models = await globModels(
               entity,
-              baseDirOrOptions.baseDir,
+              initOptions.baseDir,
               this.environmentService?.getModuleLoadType()
             );
             for (const model of models) {
@@ -84,11 +93,11 @@ export abstract class DataSourceManager<
             this.modelMapping.set(entity, dataSourceName);
           }
         }
-        (dataSourceOptions[baseDirOrOptions.entitiesConfigKey] as any) =
+        (dataSourceOptions[initOptions.entitiesConfigKey] as any) =
           Array.from(entities);
         debug(
           `[core]: DataManager load ${
-            dataSourceOptions[baseDirOrOptions.entitiesConfigKey].length
+            dataSourceOptions[initOptions.entitiesConfigKey].length
           } models from ${dataSourceName}.`
         );
       }
@@ -98,6 +107,15 @@ export abstract class DataSourceManager<
         validateConnection: dataSourceConfig.validateConnection,
       };
       await this.createInstance(dataSourceOptions, dataSourceName, opts);
+    };
+
+    const dataSourceNames = Object.keys(dataSourceConfig.dataSource);
+    if (concurrent) {
+      await Promise.all(dataSourceNames.map(processDataSource));
+    } else {
+      for (const dataSourceName of dataSourceNames) {
+        await processDataSource(dataSourceName);
+      }
     }
   }
 

packages/core/src/config/config.default.ts

@@ -6,6 +6,7 @@ export default (appInfo: MidwayAppInfo): MidwayCoreDefaultConfig => {
   return {
     core: {
       healthCheckTimeout: 1000,
+      concurrentDataSourceInitialization: false,
     },
     asyncContextManager: {
       enable: false,

packages/core/src/interface.ts

@@ -456,6 +456,7 @@ export interface MidwayCoreDefaultConfig {
   };
   core?: {
     healthCheckTimeout?: number;
+    concurrentDataSourceInitialization?: boolean;
   }
 }
 

packages/core/test/common/dataSourceManager.test.ts

@@ -3,6 +3,10 @@ import { globModels, formatGlobString } from '../../src/common/dataSourceManager
 import { join } from 'path';
 import * as assert from 'assert';
 
+function sleep(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}
+
 describe('test/common/dataSourceManager.test.ts', () => {
 
   class CustomDataSourceFactory extends DataSourceManager<any> {
@@ -247,6 +251,86 @@ describe('test/common/dataSourceManager.test.ts', () => {
 
 });
 
+describe('test concurrent data source initialization', () => {
+  class TimedDataSourceFactory extends DataSourceManager<any> {
+    getName() {
+      return 'timed';
+    }
+
+    async init(options) {
+      return super.initDataSource(options, __dirname);
+    }
+
+    protected async createDataSource(config, dataSourceName: string): Promise<any> {
+      await sleep(100);
+      return {
+        ...config,
+        dataSourceName,
+        createdAt: Date.now(),
+      };
+    }
+
+    protected async checkConnected(): Promise<boolean> {
+      return true;
+    }
+
+    protected async destroyDataSource(): Promise<void> {
+      return;
+    }
+  }
+
+  it('should initialize data sources serially by default', async () => {
+    const instance = new TimedDataSourceFactory();
+    const startTime = Date.now();
+
+    await instance.init({
+      dataSource: {
+        ds1: {},
+        ds2: {},
+        ds3: {},
+      },
+    });
+
+    const dataSources = Array.from(instance.getAllDataSources().values());
+    const creationTimes = dataSources.map(ds => ds.createdAt);
+
+    for (let i = 1; i < creationTimes.length; i++) {
+      expect(creationTimes[i] - creationTimes[i - 1]).toBeGreaterThanOrEqual(90);
+    }
+
+    expect(Date.now() - startTime).toBeGreaterThanOrEqual(290);
+  });
+
+  it('should initialize data sources concurrently when core config is enabled', async () => {
+    const instance = new TimedDataSourceFactory();
+    const startTime = Date.now();
+    instance['configService'] = {
+      getConfiguration(configKey: string) {
+        if (configKey === 'core.concurrentDataSourceInitialization') {
+          return true;
+        }
+      },
+    } as any;
+
+    await instance.init({
+      dataSource: {
+        ds1: {},
+        ds2: {},
+        ds3: {},
+      },
+    });
+
+    const dataSources = Array.from(instance.getAllDataSources().values());
+    const creationTimes = dataSources.map(ds => ds.createdAt);
+
+    for (let i = 1; i < creationTimes.length; i++) {
+      expect(Math.abs(creationTimes[i] - creationTimes[i - 1])).toBeLessThan(50);
+    }
+
+    expect(Date.now() - startTime).toBeLessThan(220);
+  });
+});
+
 describe('test global pattern', () => {
   it('should test parse global string', function () {
     expect(formatGlobString('/entity')).toEqual([

packages/info/test/__snapshots__/index.test.ts.snap

@@ -4,7 +4,7 @@ exports[`test/index.test.ts should test get config and filter secret 1`] = `
 {
   "info": {
     "asyncContextManager": "{"enable":false}",
-    "core": "{"healthCheckTimeout":1000}",
+    "core": "{"healthCheckTimeout":1000,"concurrentDataSourceInitialization":false}",
     "cos": "{"default":{"SecretId":"df******af","SecretKey":"te******vc","SecurityToken":"fj*********af"}}",
     "debug": "{"recordConfigMergeOrder":true}",
     "info": "{"title":"Midway Info","infoPath":"/_info","hiddenKey":["k***","****","*****n","********","p****"],"ignoreKey":[]}",