<commit_before>
var through = require('through2');

function filterTypeStream( type ){
  var stream = through.obj( function( data, enc, next ) {

    // data is an array of items
    if( Array.isArray( data ) ){
      data.forEach( function( item ){
        if( item.type === type ){
          this.push( item );
        }
      }, this);
    }

    // data is a single item
    else {
      if( data.type === type ){
        this.push( data );
      }
    }

    next();

  });
  return stream;
}

module.exports = function( streams ){
  if( !streams || !streams.node || !streams.way || !streams.relation ){
    throw new Error( 'invalid types stream constructor' );
  }

  var node_stream = filterTypeStream( 'node' );
  node_stream.pipe( streams.node );

  var way_stream = filterTypeStream( 'way' );
  way_stream.pipe( streams.way );

  // var seenWays = false;
  // way_stream.pipe( through.obj( function( i, e, n ){
  //   if( !seenWays ){
  //     streams.node.end();
  //     seenWays = true;
  //   }
  //   n();
  // }));

  var relation_stream = filterTypeStream( 'relation' );
  relation_stream.pipe( streams.relation );

  // simple passthrough stream
  var stream = through.obj( function( data, enc, next ) {
    this.push( data );
    next();
  });

  // pipe to each filter stream
  stream.pipe( node_stream );
  stream.pipe( way_stream );
  stream.pipe( relation_stream );
  
  // catch stream errors
  stream.on( 'error', console.error.bind( console, __filename ) );

  return stream;
};<commit_msg>remove comments<commit_after>
var through = require('through2');

function filterTypeStream( type ){
  var stream = through.obj( function( data, enc, next ) {

    // data is an array of items
    if( Array.isArray( data ) ){
      data.forEach( function( item ){
        if( item.type === type ){
          this.push( item );
        }
      }, this);
    }

    // data is a single item
    else {
      if( data.type === type ){
        this.push( data );
      }
    }

    next();

  });
  return stream;
}

module.exports = function( streams ){
  if( !streams || !streams.node || !streams.way || !streams.relation ){
    throw new Error( 'invalid types stream constructor' );
  }

  var node_stream = filterTypeStream( 'node' );
  node_stream.pipe( streams.node );

  var way_stream = filterTypeStream( 'way' );
  way_stream.pipe( streams.way );

  var relation_stream = filterTypeStream( 'relation' );
  relation_stream.pipe( streams.relation );

  // simple passthrough stream
  var stream = through.obj( function( data, enc, next ) {
    this.push( data );
    next();
  });

  // pipe to each filter stream
  stream.pipe( node_stream );
  stream.pipe( way_stream );
  stream.pipe( relation_stream );
  
  // catch stream errors
  stream.on( 'error', console.error.bind( console, __filename ) );

  return stream;
};
{'fraction_non_alphanumeric': 0.14637579169598874, 'fraction_numerical': 0.0007037297677691766, 'mean_word_length': 2.571608040201005, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3264054', 'n_tokens_mistral': 904, 'n_tokens_neox': 886, 'n_words': 242}