7
\$\begingroup\$

I'm using io.js. I have to read some files as a stream, process them with the same function, and join them back into a stream. I decided to create my own solution with Transform streams and PassThrough streams.

  "use strict";

  let fs=require('fs');
  let filePaths=['./tmp/h.txt','./tmp/s.txt'];

  let Stream = require('stream');

   class StreamProcessor  {

        constructor() {
       this.process_streams = [];
       }

       push (source_stream) { 
         // Create a new Transform Stream
        let transform = new StreamTransformer();
        // Register the finish event and then pipe
        transform.processed = transform.wait.call(transform);
          source_stream.pipe(transform);
       // push the stream to the internal array
         this.process_streams.push(transform);
     }

       done (callback) {  
         let streams = this.process_streams;
       // Wait for all Transform streams to finish processing
        Promise.all( 
          streams.map(function(s) {return s.processed;}  )
         )
       .then ( function() {
          let combined_stream=new Stream.PassThrough();
         streams.forEach(function (stream) {
         stream.pipe(combined_stream); 
       });

       // Call the callback with stream or error
       callback(null,combined_stream);
       })
        .catch(function (err) {
        callback(err);
        });
       }

     }     

      class StreamTransformer extends Stream.Transform {

          constructor () {
           // call super
          super();
         } 

         _transform(chunk,enc, transformed) {
           // process files here 
              let data=chunk.toString();
               data=data.substring(0,data.length-2);
                this.push(data);
                 transformed();
        }

          _flush(flushed) {
              // for additonal at end
              this.push('\n');
               flushed();
             }

       wait() {
          // returns a promise that resolves, when all the data is processed;
           let stream = this;

           return new Promise(function(resolve,reject)  {
       stream.on('finish', function() {
       resolve(true); });
      stream.on('error', function(err) {
         reject(err);  });
      });

       }

    }

         ///  Now simply  

          let process_stream = new StreamProcessor(); 

           filePaths.forEach(function (fpath) {
               let fstream = fs.createReadStream(fpath);
             process_stream.push(fstream);
         });

           process_stream.done(  function
          (err,combined_stream) {
          // Consume the combines stream
          combined_stream.pipe(process.stdout);
           });

Test files contain 'hello' and 'stream, and outputs:

     hell
     stream

I think this can be improved further.

\$\endgroup\$
  • 2
    \$\begingroup\$ Does your code work? At the bottom, you show that the output from this code does not match what the files actually contain. \$\endgroup\$ – SirPython Feb 16 '16 at 15:57
  • \$\begingroup\$ It works, he drops in his transformation the last 2 characters. \$\endgroup\$ – konijn Sep 16 '16 at 21:18
1
\$\begingroup\$

From a good look;

  • Your indentation is horrendous, it makes your question hard to review. Please do consistent indentation. Use http://jsbeautifier.org/ if you cannot be bothered.

  • It would have been good to document what the transformer does (drop 2 last characters)

  • I would have one-lined this:

     _transform(chunk,enc, transformed) {
       // process files here 
          let data=chunk.toString();
           data=data.substring(0,data.length-2);
            this.push(data);
             transformed();
    }
    

    to

    // process files here 
    transform(chunk, enc, transformed) {
      this.push(chunk.toString().slice(0,-2));
      transformed();
    }
    
  • If you are going to call in a handler 1 function with a signature that matches the handler like here:

    stream.on('error', function(err) {
      reject(err);  });
    });
    

    then consider to use that function as the callback:

    stream.on('error', reject);
    
  • When I take a step back, then I don't like the approach. You have this modern approach of the stream which is flowing data. Then you take the stream, plug it until your entire transformation is done, and then let it flow again. I might actually try to rewrite this.

I rewrote all that code, this is how streams should work: pipe your file into your transform and pipe that to stdout.

"use strict";

let fs=require('fs'),
    filePaths=['./smurf.js','./gargamel.js'];

const Stream = require('stream');
const Transform = require('stream').Transform;

const myTransform = new Transform({
  transform(chunk, encoding, callback) {
    callback( null, chunk.toString().slice(0,-2) ); 
  }
});

filePaths.forEach(function (path) {
  fs.createReadStream(path).pipe( myTransform ).pipe( process.stdout );
});
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.