<commit_before>"use strict";

var redis = require('redis');
var _ = require('underscore')._;

var eventSplitter = /\s+/;

var Tap = function(opts) {

  var self = this;

  if(!opts){
    opts = {};
  }

  this.opts = opts;

  if(!opts.hasOwnProperty('redis')){
    opts.redis = {
      host: '127.0.0.1',
      port: 6379
    };
  }

  if(!opts.redis.hasOwnProperty('prefix')){
    opts.redis.prefix = 'straw';
  }

  this.clients = {
    input: redis.createClient(opts.redis),
    output: redis.createClient(opts.redis)
  };

  if ( opts.input === undefined ) {
    opts.input = [];
  }

  if ( opts.inputs === undefined ) {
    opts.inputs = [];
  }

  if ( opts.output === undefined ) {
    opts.output = [];
  } else if (typeof opts.output !== 'undefined' ) {
    if ( ! Array.isArray(opts.output)) {
      opts.output = [opts.output];
    }
  }

  if ( opts.outputs === undefined ) {
    opts.outputs = {};
  }


  this.send = function() {    
    
    var outlet = false;
    var message;
    var done = function(){};

    // output a message, optionally on a named outlet
    //
    // output ([outlet,] message [, done])

    if ( arguments.length === 1 ) {
      message = arguments[0];
    }

    if ( arguments.length === 2 ) {
      // message, done
      if ( typeof arguments[1] === 'function' ){
        message = arguments[0];
        done = arguments[1];
      } else {
        outlet = arguments[0];
        message = arguments[1];
      }
    }

    if ( arguments.length === 3 ) {
      outlet = arguments[0];
      message = arguments[1];
      done = arguments[2];
    }

    if (!outlet) {     
      // no outlet specified - send out the default(s)
      self.opts.output.forEach(function(x){
        self.clients.output.lpush(opts.redis.prefix + ':' + x, JSON.stringify(message) );
      });
      return done(false);
    }

    if (self.opts.outputs.hasOwnProperty(outlet)) {
      self.clients.output.lpush(opts.redis.prefix + ':' + self.opts.outputs[outlet], JSON.stringify(message) );
      return done(false);
    }

  };

  this.brpop = function() {
    var self = this;
    self.clients.input.brpop(self.lkeys, function(err, reply){      
      if (!reply){        
        // brpop timed out
        self.brpop();
        return;          
      }
      var message = false;
      if(!err){
        try {
          message = JSON.parse(reply[1]);
        } catch (e) {
          // bad json
          self.brpop();
          return;          
        }
        self.emit('message', message);
        self.brpop();
      }
    });
  };

  // set up pipes then run the node

  // if only one input, arrayize it. if multiple inputs are provided,
  // they will already be in an array
  if (!Array.isArray(opts.input)) {
    opts.input = [opts.input];
  }

  this.lkeys = [];
  opts.input.forEach(function(x){
    self.lkeys.push(opts.redis.prefix + ':' + x);
  });

  // last element is polling timeout
  if (this.lkeys.length > 0) {    
    this.lkeys.push(60);
    // start pulling data from the pipes
    this.brpop();
  }

};
<commit_msg>Reduced polling timeout<commit_after>"use strict";

var redis = require('redis');
var _ = require('underscore')._;

var eventSplitter = /\s+/;

var Tap = function(opts) {

  var self = this;

  if(!opts){
    opts = {};
  }

  this.opts = opts;

  if(!opts.hasOwnProperty('redis')){
    opts.redis = {
      host: '127.0.0.1',
      port: 6379
    };
  }

  if(!opts.redis.hasOwnProperty('prefix')){
    opts.redis.prefix = 'straw';
  }

  this.clients = {
    input: redis.createClient(opts.redis),
    output: redis.createClient(opts.redis)
  };

  if ( opts.input === undefined ) {
    opts.input = [];
  }

  if ( opts.inputs === undefined ) {
    opts.inputs = [];
  }

  if ( opts.output === undefined ) {
    opts.output = [];
  } else if (typeof opts.output !== 'undefined' ) {
    if ( ! Array.isArray(opts.output)) {
      opts.output = [opts.output];
    }
  }

  if ( opts.outputs === undefined ) {
    opts.outputs = {};
  }


  this.send = function() {    
    
    var outlet = false;
    var message;
    var done = function(){};

    // output a message, optionally on a named outlet
    //
    // output ([outlet,] message [, done])

    if ( arguments.length === 1 ) {
      message = arguments[0];
    }

    if ( arguments.length === 2 ) {
      // message, done
      if ( typeof arguments[1] === 'function' ){
        message = arguments[0];
        done = arguments[1];
      } else {
        outlet = arguments[0];
        message = arguments[1];
      }
    }

    if ( arguments.length === 3 ) {
      outlet = arguments[0];
      message = arguments[1];
      done = arguments[2];
    }

    if (!outlet) {     
      // no outlet specified - send out the default(s)
      self.opts.output.forEach(function(x){
        self.clients.output.lpush(opts.redis.prefix + ':' + x, JSON.stringify(message) );
      });
      return done(false);
    }

    if (self.opts.outputs.hasOwnProperty(outlet)) {
      self.clients.output.lpush(opts.redis.prefix + ':' + self.opts.outputs[outlet], JSON.stringify(message) );
      return done(false);
    }

  };

  this.brpop = function() {
    var self = this;
    self.clients.input.brpop(self.lkeys, function(err, reply){      
      if (!reply){        
        // brpop timed out
        self.brpop();
        return;          
      }
      var message = false;
      if(!err){
        try {
          message = JSON.parse(reply[1]);
        } catch (e) {
          // bad json
          self.brpop();
          return;          
        }
        self.emit('message', message);
        self.brpop();
      }
    });
  };

  // set up pipes then run the node

  // if only one input, arrayize it. if multiple inputs are provided,
  // they will already be in an array
  if (!Array.isArray(opts.input)) {
    opts.input = [opts.input];
  }

  this.lkeys = [];
  opts.input.forEach(function(x){
    self.lkeys.push(opts.redis.prefix + ':' + x);
  });

  // last element is polling timeout
  if (this.lkeys.length > 0) {    
    this.lkeys.push(1);
    // start pulling data from the pipes
    this.brpop();
  }

};

{'fraction_non_alphanumeric': 0.1612221680481066, 'fraction_numerical': 0.008288639687957094, 'mean_word_length': 1.9686444766039557, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '6106127', 'n_tokens_mistral': 2084, 'n_tokens_neox': 1999, 'n_words': 432}