import { log } from "../util/logging" ; import { ITicks , INeonNotification , IMergerItemOptions , INeonTransferChunk , IChunk , } from "../util/progress" ; import { NativeComputationManager } from "../native_computation_manager" ; import { RustConcatenatorChannel , RustMergerChannel } from "../native" ; import { CancelablePromise } from '../util/promise' ; export interface IMergeParams { configFile : string ; out : string ; chunk_size ? : number ; append : boolean ; stdout : boolean ; statusUpdates : boolean ; } export interface ConcatenatorInput { path : string ; tag : string ; } export interface IConcatFilesParams { configFile : string ; out : string ; append : boolean ; } export interface IMergeFilesOptions { append ? : boolean ; chunk_size ? : number ; } export interface IMergeFilesOptionsChecked { append : boolean ; chunk_size : number ; } export type TMergeFilesEvents = 'result' | 'progress' | 'notification' ; export type TMergeFilesEventResult = ( event : IChunk ) => void ; export type TMergeFilesEventProgress = ( event : ITicks ) => void ; export type TMergeFilesEventNotification = ( event : INeonNotification ) => void ; export type TMergeFilesEventObject = TMergeFilesEventResult | TMergeFilesEventProgress | TMergeFilesEventNotification ; export function mergeFilesAsync ( config , outFile , options ? ) { return new CancelablePromise < void , void , TMergeFilesEvents , TMergeFilesEventObject > ( ( resolve , reject , cancel , refCancelCB , self ) => { try { const opt = getDefaultMergeOptions ( options ) ; log ( `mergeFilesAsync called with config: ${ JSON . stringify ( config ) }` ) ; const channel = new RustMergerChannel ( config , outFile , opt . append , opt . chunk_size ) ; const computation = new NativeComputationManager < INeonTransferChunk > ( channel ) ; refCancelCB ( ( ) => { log ( `Get command "break" operation. Starting breaking.` ) ; computation . requestShutdown ( ) ; } ) ; let totalTicks = 1 ; computation . onItem ( ( c ) => { self . emit ( 'result' , { bytesStart : c . b [ 0 ] , bytesEnd : c . b [ 1 ] , rowsStart : c . r [ 0 ] , rowsEnd : c . r [ 1 ] , } ) ; } ) ; computation . onProgress ( ( ticks ) => { totalTicks = ticks . total ; self . emit ( 'progress' , ticks ) ; } ) ; computation . onError ( ( ) => { log ( `Event "Error" is triggered` ) ; computation . requestShutdown ( ) ; } ) ; computation . onStopped ( ( ) => { log ( `Event "Stopped" is triggered` ) ; cancel ( ) ; } ) ; computation . onNotification ( ( notification ) => { self . emit ( 'notification' , notification ) ; } ) ; computation . onFinished ( ( ) => { self . emit ( 'progress' , { ellapsed : totalTicks , total : totalTicks , } ) ; resolve ( ) ; } ) ; } catch ( err ) { if ( ! ( err instanceof Error ) ) { log ( `operation is stopped. Error isn't valid:` ) ; log ( err ) ; err = new Error ( `operation is stopped. Error isn't valid.` ) ; } else { log ( `operation is stopped due error: ${ err . message }` ) ; } reject ( err ) ; } } ) ; } export type TConcatFilesEvents = 'result' | 'progress' | 'notification' ; export type TConcatFilesEventResult = ( event : IChunk ) => void ; export type TConcatFilesEventProgress = ( event : ITicks ) => void ; export type TConcatFilesEventNotification = ( event : INeonNotification ) => void ; export type TConcatFilesEventObject = TConcatFilesEventResult | TConcatFilesEventProgress | TConcatFilesEventNotification ; param config param outFile param options export function concatFilesAsync ( config , outFile , options ? ) { return new CancelablePromise < void , void , TConcatFilesEvents , TConcatFilesEventObject > ( ( resolve , reject , cancel , refCancelCB , self ) => { try { const opt = getDefaultMergeOptions ( options ) ; const channel = new RustConcatenatorChannel ( config , outFile , opt . append , opt . chunk_size ) ; const computation = new NativeComputationManager < INeonTransferChunk > ( channel ) ; refCancelCB ( ( ) => { log ( `Get command "break" operation. Starting breaking.` ) ; computation . requestShutdown ( ) ; } ) ; let totalTicks = 1 ; computation . onItem ( ( c ) => { self . emit ( 'result' , { bytesStart : c . b [ 0 ] , bytesEnd : c . b [ 1 ] , rowsStart : c . r [ 0 ] , rowsEnd : c . r [ 1 ] , } ) ; } ) ; computation . onProgress ( ( ticks ) => { totalTicks = ticks . total ; self . emit ( 'progress' , ticks ) ; } ) ; computation . onError ( ( ) => { log ( `Event "Error" is triggered` ) ; computation . requestShutdown ( ) ; } ) ; computation . onStopped ( ( ) => { log ( `Event "Stopped" is triggered` ) ; cancel ( ) ; } ) ; computation . onNotification ( ( notification ) => { self . emit ( 'notification' , notification ) ; } ) ; computation . onFinished ( ( ) => { self . emit ( 'progress' , { ellapsed : totalTicks , total : totalTicks , } ) ; resolve ( ) ; } ) ; } catch ( err ) { if ( ! ( err instanceof Error ) ) { log ( `operation is stopped. Error isn't valid:` ) ; log ( err ) ; err = new Error ( `operation is stopped. Error isn't valid.` ) ; } else { log ( `operation is stopped due error: ${ err . message }` ) ; } reject ( err ) ; } } ) ; } function getDefaultMergeOptions ( options ) { if ( typeof options !== 'object' || options === null ) { options = { } ; } options . append = typeof options . append === 'boolean' ? options . append : true ; options . chunk_size = typeof options . chunk_size === 'number' ? options . chunk_size : 5000 ; return options as IMergeFilesOptionsChecked ; }