Fixes to scripts so they all consistently pass a single object through the stream at a time.

Author: ashleydavis

listing-5.js

@@ -36,21 +36,13 @@ function transformRow (inputRow) {
     return outputRow;
 };
 
-//
-// Transform a data set (in this case a chunk of rows).
-//
-function transformData (inputData) {
-    // Your code here to transform a batch of rows.
-    return inputData.map(transformRow);
-};
-
 //
 // Create a stream that converts the temperature for all records that pass through the stream.
 //
 function convertTemperatureStream () {
     const transformStream = new stream.Transform({ objectMode: true }); // Create a bidirectional stream in 'object mode'.
     transformStream._transform = (inputChunk, encoding, callback) => { // Callback to execute on chunks that are input.
-        var outputChunk = transformData(inputChunk); // Transform the chunk.
+        var outputChunk = transformRow(inputChunk); // Transform the chunk.
         transformStream.push(outputChunk); // Pass the converted chunk to the output stream.
         callback();
     };

listing-8.js

@@ -36,21 +36,14 @@ function transformRow (inputRow) {
     return outputRow;
 };
 
-//
-// Transform a data set (in this case a chunk of rows).
-//
-function transformData (inputData) {
-    // Your code here to transform a batch of rows.
-    return inputData.map(transformRow);
-};
-
 //
 // Create a stream that converts the temperature for all records that pass through the stream.
 //
 function convertTemperatureStream () {
     const transformStream = new stream.Transform({ objectMode: true }); // Create a bidirectional stream in 'object mode'.
     transformStream._transform = (inputChunk, encoding, callback) => { // Callback to execute on chunks that are input.
-        var outputChunk = transformData(inputChunk); // Transform the chunk.
+        console.log(inputChunk); //fio:
+        var outputChunk = transformRow(inputChunk); // Transform the chunk.
         transformStream.push(outputChunk); // Pass the converted chunk to the output stream.
         callback();
     };

toolkit/open-csv-input-stream.js

@@ -21,7 +21,9 @@ function openCsvInputStream (inputFilePath) {
         skipEmptyLines: true, 
 
         step: (results) => { // Handles incoming rows of CSV data.
-            csvInputStream.push(results.data); // Push results as they are streamed from the file.
+            for (let row of results.data) {
+                csvInputStream.push(row); // Push results as they are streamed from the file.                
+            }
         },
 
         complete: () => { // File read operation has completed.

toolkit/open-csv-output-stream.js

@@ -14,7 +14,7 @@ function openCsvOutputStream (outputFilePath) {
 
     const csvOutputStream = new stream.Writable({ objectMode: true }); // Create stream for writing data records, note that 'object mode' is enabled.
     csvOutputStream._write = (chunk, encoding, callback) => { // Handle writes to the stream.
-        var outputCSV = papaparse.unparse(chunk, { 
+        var outputCSV = papaparse.unparse([chunk], { 
             header: firstOutput
         });
         fileOutputStream.write(outputCSV + '\n');

toolkit/open-json-output-stream.js

@@ -16,13 +16,13 @@ function openJsonOutputStream (outputFilePath) {
     const jsonOutputStream = new stream.Writable({ objectMode: true });
     jsonOutputStream._write = (chunk, encoding, callback) => {
         if (numRecords > 0) {
-            fileOutputStream.write(",");
+            fileOutputStream.write(",\n");
         }
 
         // Output a single row of a JSON array.
-        const jsonData = JSON.stringify(curObject);
+        const jsonData = JSON.stringify(chunk);
         fileOutputStream.write(jsonData);
-        numRecords += chunk.length;
+        ++numRecords;
         callback();        
     };