NODE-111 : Upgrade Node-Client-Api to support an upgraded NODE 18

Author: anu3990

examples/readAll-documents.js

@@ -33,11 +33,11 @@ for(let i=0; i<100; i++) {
 }
 uriStream.push(null);
 
-uriStream.pipe(dbWriter.documents.readAll({
+dbWriter.documents.readAll(uriStream,{
     categories: ['permissions', 'metadataValues', 'collections', 'quality'],
     onCompletion: ((summary) => {
         console.log(summary.docsReadSuccessfully+' documents were read successfully.');
         console.log(summary.docsFailedToBeRead+' documents failed to be read.');
         console.log('Time taken was '+summary.timeElapsed+' milliseconds.');
     })
-}));
+});

examples/removeAllUris-documents.js

@@ -29,11 +29,11 @@ for(let i=0; i<100; i++) {
 }
 removeAllUrisStream.push(null);
 
-removeAllUrisStream.pipe(dbWriter.documents.removeAllUris({
+dbWriter.documents.removeAllUris(removeAllUrisStream,{
     concurrentRequests : {multipleOf:'hosts', multiplier:4},
     onCompletion: ((summary) => {
         console.log(summary.docsRemovedSuccessfully+' documents were removed successfully.');
         console.log(summary.docsFailedToBeRemoved+' documents failed to be removed.');
         console.log('Time taken was '+summary.timeElapsed+' milliseconds.');
     })
-}));
+});

examples/transformAll-documents.js

@@ -31,11 +31,11 @@ for(let i=0; i<100; i++) {
 }
 transformStream.push(null);
 
-transformStream.pipe(dbWriter.documents.transformAll({
+dbWriter.documents.transformAll(transformStream,{
     transform: [transformName, {'field_in_the_document':'Transformation_needed'}],
     onCompletion: ((summary) => {
         console.log(summary.docsTransformedSuccessfully+' documents were transformed successfully.');
         console.log(summary.docsFailedToBeTransformed+' documents failed to be transformed.');
         console.log('Time taken was '+summary.timeElapsed+' milliseconds.');
     })
-}));
+});

examples/writeAll-documents.js

@@ -32,7 +32,7 @@ for(let i=0; i<1000; i++) {
 }
 readable.push(null);
 
-readable.pipe(dbWriter.documents.writeAll({
+dbWriter.documents.writeAll(readable,{
 
     onCompletion: ((summary) => {
         console.log(summary.docsWrittenSuccessfully+' documents were written successfully.');
@@ -46,4 +46,4 @@ readable.pipe(dbWriter.documents.writeAll({
         console.log('Time taken was '+progressSoFar.timeElapsed+' milliseconds.');
         return null;
     })
-}));
+});

lib/documents.js

@@ -2260,6 +2260,7 @@ Documents.prototype.suggest = function suggestDocuments() {
  *  defaultMetadata and transform.
  * @method documents#writeAll
  * @since 2.8.0
+ * @param {stream} [stream] - stream containing the input.
  * @param {object} [options] - Configures the write operation.
  * @param {function(progress, documents)} [onBatchSuccess] - A callback that can notify other systems about batches
  *  written successfully.
@@ -2275,18 +2276,17 @@ Documents.prototype.suggest = function suggestDocuments() {
  * @returns {stream.Writable} - a stream.Writable in object mode that receives document descriptor input from the
  *  application for writing to the database.
  */
-Documents.prototype.writeAll = function writeAllDocuments(options) {
+Documents.prototype.writeAll = function writeAllDocuments(stream,options) {
   return writeAllDocumentsImpl.call(
-      this, options
+      this, stream,options
   );
 };
-function writeAllDocumentsImpl(jobOptions) {
+function writeAllDocumentsImpl(inputStream,jobOptions) {
 
   let path = '/v1/internal/forestinfo';
   let connectionParams = this.client.getConnectionParams();
   let requestOptions = mlutil.copyProperties(connectionParams);
 
-  let inputStream = new stream.PassThrough({objectMode: true});
   requestOptions.method = 'GET';
   requestOptions.headers = {
     'Accept':       'application/json',
@@ -2345,7 +2345,7 @@ function writeAllDocumentsImpl(jobOptions) {
 
 function finishWriter(jobState) {
   jobState.requesterCount--;
-  if(jobState.requesterCount === 0) {
+  if(jobState.requesterCount <= 0) {
     if(jobState.jobOptions.onCompletion){
       const summary = {
         docsWrittenSuccessfully: jobState.docsWrittenSuccessfully,
@@ -2380,6 +2380,7 @@ function onWriteAllInit(output) {
     }
   }
   jobState.startTime = Date.now();
+  jobState.requesterCount =0;
   for (let i = 0; i < maxRequesters; i++) {
     jobState.requesterCount++;
     onWriteAllDocs(jobState,i);
@@ -2765,6 +2766,7 @@ function urisOutputTransform(headers, data) {
  *  concurrentRequests, categories, transform, onInitialTimestamp, outputStreamType and consistentSnapshot.
  * @method documents#readAll
  * @since 2.9.0
+ * @param {stream} [stream] - stream containing the input uris.
  * @param {object} [options] - Configures the readAll operation.
  * @param {function(summary)} [onCompletion] - A callback that receives a summary of the results.
  * @param {function(progress, uris, error)} [onBatchError] - A callback that responds  to any error while reading a
@@ -2791,13 +2793,14 @@ function urisOutputTransform(headers, data) {
  *  object mode (for arrays of strings) and returns document descriptors with the content and/or document uri as output to
  *  the application in object mode.
  */
-Documents.prototype.readAll = function readAllDocuments(options) {
+Documents.prototype.readAll = function readAllDocuments(stream, options) {
+
   return readAllDocumentsImpl.call(
-      this, options
+      this, stream, options
   );
 };
 
-function readAllDocumentsImpl(jobOptions) {
+function readAllDocumentsImpl(inputStream, jobOptions) {
 
   let path = '/v1/internal/forestinfo';
   let connectionParams = this.client.getConnectionParams();
@@ -2829,8 +2832,7 @@ function readAllDocumentsImpl(jobOptions) {
       throw new Error('Invalid value for inputKind. Value must be array or string.');
     }
   }
-
-  jobState.inputStream = new stream.PassThrough({objectMode: true});
+  jobState.inputStream = inputStream?inputStream:new stream.PassThrough({objectMode: true});
 
   if(jobState.jobOptions.outputStreamType){
     readableObjectMode = jobState.jobOptions.outputStreamType.toString().toLowerCase();
@@ -3242,7 +3244,7 @@ function queryToReadAllDocumentsImpl(query,jobOptions){
     if(!pipelined) {
       pipelined = true;
 
-      const readAllStream = readAllDocumentsImpl.call(docInstance, readAllJobOptions);
+      const readAllStream = readAllDocumentsImpl.call(docInstance,null, readAllJobOptions);
       readAllStream.on('error', function(err){
         queryToReadAllResultStream.emit(err);
       });
@@ -3353,7 +3355,8 @@ function copyConstraints(suggestConstraints, searchConstraints) {
  * The options include onCompletion, batchSize, onBatchError, inputKind, concurrentRequests,  transform, onBatchSuccess
  *   and transformStrategy.
  * @method documents#transformAll
- * @since 2.10.0
+ * @since 3.0.0
+ * @param {stream} [stream] - stream containing the input.
  * @param {object} [options] - Configures the transformAll operation.
  * @param {function(summary)} [onCompletion] - A callback that receives a summary of the results.
  * @param {function(progress, uris, error)} [onBatchError] - A callback that responds  to any error while transforming a
@@ -3375,13 +3378,13 @@ function copyConstraints(suggestConstraints, searchConstraints) {
  * @returns {stream.Writable} - returns a writable stream that takes the document uris of the database documents to be
  *  transformed.
  */
-Documents.prototype.transformAll = function transformAllDocuments(options) {
+Documents.prototype.transformAll = function transformAllDocuments(stream, options) {
   return transformAllDocumentsImpl.call(
-      this, options
+      this, stream, options
   );
 };
 
-function transformAllDocumentsImpl(jobOptions){
+function transformAllDocumentsImpl(inputStream, jobOptions){
   let path = '/v1/internal/forestinfo';
   let connectionParams = this.client.getConnectionParams();
   let requestOptions = mlutil.copyProperties(connectionParams);
@@ -3403,7 +3406,7 @@ function transformAllDocumentsImpl(jobOptions){
     docsFailedToBeTransformed: 0,
     jobOptions: (jobOptions)? mlutil.copyProperties(jobOptions):{}
   };
-
+  jobState.stream = inputStream?inputStream:new stream.PassThrough({objectMode: true});
   if(!jobState.jobOptions.transform){
     throw new Error('transform name needed while using transformAll api');
   }
@@ -3416,12 +3419,9 @@ function transformAllDocumentsImpl(jobOptions){
     if(inputKindValue === 'array' && jobState.jobOptions.batchSize){
       throw new Error('batchSize not expected when inputKind is array.');
     }
-    jobState.stream = (inputKindValue === 'string') ? new stream.PassThrough():
-        new stream.PassThrough({objectMode: true});
     jobState.inputKindValue = inputKindValue;
   }
   else {
-    jobState.stream = new stream.PassThrough({objectMode: true});
     jobState.inputKindValue = 'string';
   }
 
@@ -3674,7 +3674,7 @@ function finishTransformAll(jobState) {
  * The queryToTransformAll is a convenience function that combines the queryAll and transformAll operations.
  * The transform used to configure the documents must be installed on the server beforehand.
  * @method documents#queryToTransformAll
- * @since 2.10.0
+ * @since 3.0.0
  * @param {ctsQuery} query - A query built by the ctsQueryBuilder.
  * @param {object} [options] - Configures the queryToTransformAll operation.
  * @param {boolean|DatabaseClient.timestamp} [consistentSnapshot] - Controls whether to get an immutable view of the result set.
@@ -3765,7 +3765,7 @@ function queryToTransformAllDocumentsImpl(query, jobOptions){
   queryAllStream.on('data', function(item){
     if(!pipelined) {
       pipelined = true;
-      const transformAllStream = transformAllDocumentsImpl.call(docInstance, transformAllJobOptions);
+      const transformAllStream = transformAllDocumentsImpl.call(docInstance,null, transformAllJobOptions);
 
       queryAllStream.on('error', function(err){
         transformAllStream.emit(err);
@@ -3779,7 +3779,8 @@ function queryToTransformAllDocumentsImpl(query, jobOptions){
 /**
  * The removeAllUris function deletes a set of documents from the database.
  * @method documents#removeAllUris
- * @since 2.10.0
+ * @since 3.0.0
+ * @param {stream} [stream] - stream containing the input uris.
  * @param {object} [options] - Configures the removeAllUris operation.
  * @param {function(summary)} [onCompletion] - A callback that receives a summary of the results.
  * @param {function(progress, uris, error)} [onBatchError] - A callback that responds  to any error while transforming a
@@ -3795,13 +3796,13 @@ function queryToTransformAllDocumentsImpl(query, jobOptions){
  * @returns {stream.Writable} - a stream.Writable that receives document URI input from the application in string mode
  *  or (for arrays of strings) object mode.
  */
-Documents.prototype.removeAllUris = function removeAllUrisDocuments(options) {
+Documents.prototype.removeAllUris = function removeAllUrisDocuments(stream, options) {
   return removeAllUrisDocumentsImpl.call(
-      this, options
+      this, stream, options
   );
 };
 
-function removeAllUrisDocumentsImpl(jobOptions){
+function removeAllUrisDocumentsImpl(inputStream, jobOptions){
   let path = '/v1/internal/forestinfo';
   let connectionParams = this.client.getConnectionParams();
   let requestOptions = mlutil.copyProperties(connectionParams);
@@ -3823,7 +3824,7 @@ function removeAllUrisDocumentsImpl(jobOptions){
     docsFailedToBeRemoved: 0,
     jobOptions: (jobOptions)? mlutil.copyProperties(jobOptions):{}
   };
-
+  jobState.stream = inputStream?inputStream:new stream.PassThrough({objectMode: true});
   if(jobState.jobOptions.inputKind){
     const inputKindValue = jobState.jobOptions.inputKind.toString().toLowerCase();
     if (!new Set(['string', 'array']).has(inputKindValue)) {
@@ -3832,12 +3833,9 @@ function removeAllUrisDocumentsImpl(jobOptions){
     if(inputKindValue === 'array' && jobState.jobOptions.batchSize){
       throw new Error('batchSize not expected when inputKind is array.');
     }
-    jobState.stream = (inputKindValue === 'string') ? new stream.PassThrough():
-        new stream.PassThrough({objectMode: true});
     jobState.inputKindValue = inputKindValue;
   }
   else {
-    jobState.stream = new stream.PassThrough({objectMode: true});
     jobState.inputKindValue = 'string';
   }
 
@@ -4049,7 +4047,7 @@ function removeDocs(jobState, removeBatchArray, writerId){
 /**
  * The queryToRemoveAll is a convenience function that combines the queryAll and removeAllUris operations.
  * @method documents#queryToRemoveAll
- * @since 2.10.0
+ * @since 3.0.0
  * @param {ctsQuery} query - A query built by the ctsQueryBuilder.
  * @param {object} [options] - Configures the queryToRemoveAll operation.
  * @param {boolean|DatabaseClient.timestamp} [consistentSnapshot] - Controls whether to get an immutable view of the result set.
@@ -4129,7 +4127,7 @@ function queryToRemoveAllDocumentsImpl(query, jobOptions){
   queryAllStream.on('data', function(item){
     if(!pipelined) {
       pipelined = true;
-      const removeAllStream = removeAllUrisDocumentsImpl.call(docInstance, removeAllJobOptions);
+      const removeAllStream = removeAllUrisDocumentsImpl.call(docInstance, null,removeAllJobOptions);
 
       queryAllStream.on('error', function(err){
         removeAllStream.emit(err);

lib/rows.js

@@ -407,7 +407,7 @@ Rows.prototype.generateView = function generateViewRows(builtPlan, schema, view,
  * The optional options include onCompletion, batchSize, onBatchError, queryType,
  *  concurrentRequests, columnTypes, rowStructure, rowFormat, onInitialTimestamp, outputStreamType and consistentSnapshot.
  * @method rows#queryAll
- * @since 2.10.0
+ * @since 3.0.0
  * @param {planBuilder.ModifyPlan|object|string} batchView - A query that exports a modified subset of rows from a view
  *  expressed as a modify plan without parameters built by the PlanBuilder, a JavaScript literal object equivalent to a
  *  JSON AST representation of the modify plan or a string literal with the JSON AST or Query DSL representation of the