engine/DataSession.js
/**
		 * Processes all the {@link DataOperation} instances held within the session.
		 *
		 * @public
		 * @param {DataProvider} dataProvider
		 * @returns {Promise}
		 */
Processes all the  DataOperation instances held within the session.
@public
@param {DataProvider} dataProvider
@returns {Promise}
Processes all the DataOperation instances held within the session . @public @param { DataProvider } dataProvider @returns { Promise }
flush(dataProvider) {
			return Promise.resolve()
				.then(() => {
					assert.argumentIsRequired(dataProvider, 'dataProvider', DataProvider, 'DataProvider');
					if (this._flushed) {
						throw new Error(`Session [ ${this._instanceCounter}  has already been flushed.`);
					}
					this._flushed = true;
					logger.info('Session [', this._instanceCounter, '] flush starting [', this._instanceId, ']');
					if (this._pending.empty()) {
						logger.warn('Session [', this._instanceCounter, '] has no operations.');
					}
					let operationCounter = 0;
					const results = [ ];
					let outputIndicies;
					if (this._resultTypes.length === 0) {
						outputIndicies = [ ];
					} else {
						outputIndicies = this._resultTypes.map(() => [ ]);
					}
					const flushRecursive = (previousResult) => {
						return Promise.resolve()
							.then(() => {
								let processPromise;
								if (this._pending.empty()) {
									processPromise = Promise.resolve(previousResult);
								} else {
									let operation = null;
									let operationCount;
									while (operation === null && !this._pending.empty()) {
										const candidate = this._pending.dequeue().operation;
										operationCount = ++operationCounter;
										if (candidate.equals(previousResult.operation)) {
											logger.debug('Session [', this._instanceCounter, '] operation [', operationCount, '][', candidate.toString() ,'] discarded as duplicate.');
										} else {
											operation = candidate;
										}
									}
									if (operation === null) {
										processPromise = Promise.resolve(previousResult);
									} else {
										this._processed.push(operation);
										logger.debug('Session [', this._instanceCounter, '] operation [', operationCount, '][', operation.toString() ,'] starting.');
										processPromise = operation.process(dataProvider, this._instanceId, this._name)
											.then((result) => {
												logger.debug('Session [', this._instanceCounter, '] operation [', operationCount, '][', operation.toString() ,'] complete.');
												results.push(result);
												const operationIndex = results.length - 1;
												if (this._resultTypes.length === 0) {
													const resultIndex = this._userEnqueued.findIndex(o => o === result.operation);
													if (!(resultIndex < 0)) {
														outputIndicies[resultIndex] = operationIndex;
													}
												} else {
													const resultIndex = this._resultTypes.findIndex(t => operation instanceof t);
													if (!(resultIndex < 0)) {
														outputIndicies[resultIndex].push(operationIndex);
													}
												}
												result.children.forEach(container => enqueue.call(this, container));
												return result;
											});
									}
									processPromise = processPromise.then((result) => {
										return flushRecursive(result);
									});
								}
								return processPromise;
							});
					};
					return flushRecursive(DataOperationResult.getInitial())
						.then(() => {
							const transformedResults = results.reduceRight((resolvedResults, result, i, a) => {
								const spawnResults = result.children.map((spawnContainer) => {
									return a.find((previousResult) => previousResult.operation === spawnContainer.operation);
								});
								resolvedResults.push(result.operation.transformResult(result, spawnResults));
								return resolvedResults;
							}, [ ]);
							const resolveOutput = (outputIndex) => {
								const reversedIndex = results.length - outputIndex - 1;
								return transformedResults[reversedIndex].result;
							};
							const output = outputIndicies.map((i) => {
								if (is.array(i)) {
									return i.map(j => resolveOutput(j));
								} else {
									return resolveOutput(i);
								}
							});
							logger.info('Session [', this._instanceCounter, '] flush finished [', this._instanceId, ']');
							if (output.length === 1) {
								return output[0];
							} else {
								return output;
							}
						});
				});
		}
flush ( dataProvider ) { return Promise . resolve ( ) . then ( ( ) => { assert . argumentIsRequired ( dataProvider , 'dataProvider' , DataProvider , 'DataProvider' ) ; if ( this . _flushed ) { throw new Error ( ` ${ this . _instanceCounter } ` ) ; } this . _flushed = true ; logger . info ( 'Session [' , this . _instanceCounter , '] flush starting [' , this . _instanceId , ']' ) ; if ( this . _pending . empty ( ) ) { logger . warn ( 'Session [' , this . _instanceCounter , '] has no operations.' ) ; } let operationCounter = 0 ; const results = [ ] ; let outputIndicies ; if ( this . _resultTypes . length === 0 ) { outputIndicies = [ ] ; } else { outputIndicies = this . _resultTypes . map ( ( ) => [ ] ) ; } const flushRecursive = ( previousResult ) => { return Promise . resolve ( ) . then ( ( ) => { let processPromise ; if ( this . _pending . empty ( ) ) { processPromise = Promise . resolve ( previousResult ) ; } else { let operation = null ; let operationCount ; while ( operation === null && ! this . _pending . empty ( ) ) { const candidate = this . _pending . dequeue ( ) . operation ; operationCount = ++ operationCounter ; if ( candidate . equals ( previousResult . operation ) ) { logger . debug ( 'Session [' , this . _instanceCounter , '] operation [' , operationCount , '][' , candidate . toString ( ) , '] discarded as duplicate.' ) ; } else { operation = candidate ; } } if ( operation === null ) { processPromise = Promise . resolve ( previousResult ) ; } else { this . _processed . push ( operation ) ; logger . debug ( 'Session [' , this . _instanceCounter , '] operation [' , operationCount , '][' , operation . toString ( ) , '] starting.' ) ; processPromise = operation . process ( dataProvider , this . _instanceId , this . _name ) . then ( ( result ) => { logger . debug ( 'Session [' , this . _instanceCounter , '] operation [' , operationCount , '][' , operation . toString ( ) , '] complete.' ) ; results . push ( result ) ; const operationIndex = results . length - 1 ; if ( this . _resultTypes . length === 0 ) { const resultIndex = this . _userEnqueued . findIndex ( o => o === result . operation ) ; if ( ! ( resultIndex < 0 ) ) { outputIndicies [ resultIndex ] = operationIndex ; } } else { const resultIndex = this . _resultTypes . findIndex ( t => operation instanceof t ) ; if ( ! ( resultIndex < 0 ) ) { outputIndicies [ resultIndex ] . push ( operationIndex ) ; } } result . children . forEach ( container => enqueue . call ( this , container ) ) ; return result ; } ) ; } processPromise = processPromise . then ( ( result ) => { return flushRecursive ( result ) ; } ) ; } return processPromise ; } ) ; } ; return flushRecursive ( DataOperationResult . getInitial ( ) ) . then ( ( ) => { const transformedResults = results . reduceRight ( ( resolvedResults , result , i , a ) => { const spawnResults = result . children . map ( ( spawnContainer ) => { return a . find ( ( previousResult ) => previousResult . operation === spawnContainer . operation ) ; } ) ; resolvedResults . push ( result . operation . transformResult ( result , spawnResults ) ) ; return resolvedResults ; } , [ ] ) ; const resolveOutput = ( outputIndex ) => { const reversedIndex = results . length - outputIndex - 1 ; return transformedResults [ reversedIndex ] . result ; } ; const output = outputIndicies . map ( ( i ) => { if ( is . array ( i ) ) { return i . map ( j => resolveOutput ( j ) ) ; } else { return resolveOutput ( i ) ; } } ) ; logger . info ( 'Session [' , this . _instanceCounter , '] flush finished [' , this . _instanceId , ']' ) ; if ( output . length === 1 ) { return output [ 0 ] ; } else { return output ; } } ) ; } ) ; }
Processes all the {@link DataOperation} instances held within the session.
Processes all the { @link DataOperation } instances held within the session .