import { ChildProcess , fork } from 'child_process' ; import { EventEmitter } from 'events' ; import * as path from 'path' ; import * as os from 'os' ; import { log , generateError } from './FsOps' ; import { processConstraintsPerTable } from './ConstraintsProcessor' ; import * as migrationStateManager from './MigrationStateManager' ; import Conversion from './Conversion' ; import MessageToDataLoader from './MessageToDataLoader' ; import MessageToMaster from './MessageToMaster' ; /**
 * A number of currently running loader processes.
 */ let loaderProcessesCount = 0 ; /**
 * "tableLoadingFinished" event.
 */ const tableLoadingFinishedEvent = 'tableLoadingFinished' ; /**
 * An EventEmitter instance.
 */ const eventEmitter = new EventEmitter ( ) ; /**
 * A path to the DataLoader.js file.
 * !!!Notice, in runtime it points to ../dist/src/DataLoader.js and not DataLoader.ts
 */ const dataLoaderPath = path . join ( __dirname , 'DataLoader.js' ) ; /**
 * Returns the options object, which intended to be used upon creation of the data loader process.
 */ const getDataLoaderOptions = ( conversion ) : any => { const options = Object . create ( null ) ; if ( conversion . _loaderMaxOldSpaceSize !== 'DEFAULT' ) { options . execArgv = [ `--max-old-space-size=${ conversion . _loaderMaxOldSpaceSize }` ] ; } return options ; } ; /**
 * Kills a process specified by the pid.
 */ const killProcess = async ( pid , conversion ) : Promise < void > => { try { process . kill ( pid ) ; } catch ( killError ) { await generateError ( conversion , `\t--[killProcess] ${ killError }` ) ; } } ; /**
 * Checks if all data chunks were processed.
 */ const dataPoolProcessed = ( conversion ) : boolean => { return conversion . _dataPool . length === 0 ; } ; /**
 * Runs the data pipe.
 */ export default ( conversion ) : Promise < Conversion > => { return new Promise < Conversion > ( resolve => { if ( dataPoolProcessed ( conversion ) ) { return resolve ( conversion ) ; } eventEmitter . on ( tableLoadingFinishedEvent , async ( tableName ) => { await processConstraintsPerTable ( conversion , tableName , conversion . shouldMigrateOnlyData ( ) ) ; if ( loaderProcessesCount === 0 ) { await migrationStateManager . set ( conversion , 'per_table_constraints_loaded' ) ; return resolve ( conversion ) ; } } ) ; const numberOfSimultaneouslyRunningLoaderProcesses = Math . min ( conversion . _dataPool . length , conversion . _maxEachDbConnectionPoolSize , os . cpus ( ) . length ) ; for ( let i = 0 ; i < numberOfSimultaneouslyRunningLoaderProcesses ; ++ i ) { runLoaderProcess ( conversion ) ; } } ) ; } ; /**
 * Runs the loader process.
 */ const runLoaderProcess = ( conversion ) : void => { if ( dataPoolProcessed ( conversion ) ) { return ; } const loaderProcess = fork ( dataLoaderPath , getDataLoaderOptions ( conversion ) ) ; loaderProcessesCount ++ ; loaderProcess . on ( 'message' , async ( signal ) => { const msg = `\t--[runLoaderProcess]  For now inserted: ${ signal . totalRowsToInsert } rows,` + `Total rows to insert into "${ conversion . _schema }"."${ signal . tableName }": ${ signal . totalRowsToInsert }` ; log ( conversion , msg ) ; await killProcess ( loaderProcess . pid , conversion ) ; loaderProcessesCount -- ; eventEmitter . emit ( tableLoadingFinishedEvent , signal . tableName ) ; runLoaderProcess ( conversion ) ; } ) ; loaderProcess . send ( new MessageToDataLoader ( conversion . _config , conversion . _dataPool . pop ( ) ) ) ; } ;