Processing files with streams 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.
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 );
});