<commit_before>var from = require('from');

module.exports = function(streams) {

  return from(function getChunk(count, next) {
    var _this = this;

    var streamQueue = [];
    var currentStream = 0;

    // Emit everything available in the queue
    function emitData() {
      for(var i=0;i<streamQueue.length;i++) {
        var dataQueue = streamQueue[i];

        if(dataQueue.length == 0) {
          return;
        }

        for(var j=0;j<dataQueue.length;j++) {
          var data = dataQueue[j];
          if(!!data) {
            _this.emit('data', data);
            dataQueue[j] = null;

            return emitData();
          }
        }
      }

      if(currentStream == streamQueue.length) {
        _this.emit('end');
      }
    }

    function processStream(index, stream) {
      stream.on('data', function(data) {
        streamQueue[index].push(data);
        emitData();
      });
      stream.on('end', function() {
        currentStream++;

        // The stream was empty and didn't send any data
        if(streamQueue[index].length == 0) {
          streamQueue[index].push(null);
        }

        //If we have received the last end event, end the whole stream.
        if(currentStream == streamQueue.length) {
          emitData();
        }
      })

      stream.resume();
    }

    for(var i=0;i<streams.length;i++) {
      streamQueue.push([]);
    }

    for(var j=0;j<streams.length;j++) {
      processStream(j, streams[j]);
    }

  });
};<commit_msg>Output info message if include directive does not match any files (Fixes #13)<commit_after>var from = require('from');
var log = require('./utils/log');
var es   = require('event-stream');

module.exports = function(streams) {

  var isEmptyStream = false;
  return from(function getChunk(count, next) {
    var _this = this;

    var streamQueue = [];
    var currentStream = 0;

    // Emit everything available in the queue
    function emitData() {
      for(var i=0;i<streamQueue.length;i++) {
        var dataQueue = streamQueue[i];

        if(dataQueue.length == 0) {
          return;
        }

        for(var j=0;j<dataQueue.length;j++) {
          var data = dataQueue[j];
          if(!!data) {
            _this.emit('data', data);
            dataQueue[j] = null;

            return emitData();
          }
        }
      }

      if(currentStream == streamQueue.length) {
        _this.emit('end');
      }
    }

    function processStream(index, stream) {
      stream.on('data', function(data) {
        streamQueue[index].push(data);
        emitData();
      });
      stream.on('end', function() {
        currentStream++;

        // The stream was empty and didn't send any data
        if(streamQueue[index].length == 0) {
          isEmptyStream = true;
          streamQueue[index].push(null);
        }

        //If we have received the last end event, end the whole stream.
        if(currentStream == streamQueue.length) {
          emitData();
        }
      })

      stream.resume();
    }

    for(var i=0;i<streams.length;i++) {
      streamQueue.push([]);
    }

    for(var j=0;j<streams.length;j++) {
      processStream(j, streams[j]);
    }

  }).pipe(es.map(function(file, cb) {
    if(isEmptyStream) {
      log('[Info] Non-matching include directive found in "' + file.path + '"');
      cb(null, file);
    }
  }));
};
{'fraction_non_alphanumeric': 0.14582714582714582, 'fraction_numerical': 0.004752004752004752, 'mean_word_length': 2.0315031503150314, 'pattern_counts': {'":': 0, '<': 11, '<?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': '484228', 'n_tokens_mistral': 1050, 'n_tokens_neox': 1010, 'n_words': 221}