<commit_before>'use strict';

var Engine = require('engine.io').Server
  , Socket = require('engine.io').Socket
  , request = require('request')
  , Route = require('routable');

/**
 * Create a new Scaler instance.
 *
 * @constructor
 * @param {Redis} redis A Redis client.
 * @param {Object} options Scaler options.
 * @api public
 */
function Scaler(redis, options) {
  options = options || {};

  //
  // The HTTP routes that we should be listening on.
  //
  this.broadcast = new Route(options.broadcast || '/scaler/broadcast');
  this.endpoint = new Route(options.endpoint || '/stream/');

  // The redis client we need to keep connection state.
  this.redis = redis || require('redis').createClient();

  // The root domain of the service, will be used for redirects.
  this.service = options.service || false;

  // The namespace for the keys that are stored in redis.
  this.namespace = options.namespace || 'scaler';

  // How long do we maintain stake from a single user.
  this.timeout = options.timeout || 60 * 15;

  // The network address this server is approachable on for internal HTTP requests.
  this.address = options.address || 'localhost';

  // The port number that we should use to connect with this server.
  this.port = options.port || null;

  //
  // Allow custom encoders and decoders, we default to JSON.parse so it should
  // have simular interfaces.
  //
  this.encode = options.encode || JSON.stringify;
  this.decode = options.decode || JSON.parse;

  //
  // These properies will be set once we're initialized.
  //
  this.server = null;         // HTTP server instance.
  this.engine = null;         // Engine.IO server.
}

//
// The scaler inherits from the EventEmitter so we can savely emit events
// without creating tail recursion.
//
Scaler.prototype.__proto__ = require('events').EventEmitter.prototype;

//
// Because we are to lazy to combine address + port every single time.
//
Object.defineProperty(Scaler.prototype, 'interface', {
  get: function get() {
    return this.address +':'+ this.port;
  }
});

/**
 * Intercept HTTP requests and handle them accordingly.
 *
 * @param {Boolean} websocket This is a WebSocket request
 * @param {Request} req HTTP request instance.
 * @param {Response} res HTTP response instance.
 * @param {Buffer} head HTTP head buffer.
 * @api private
 */
Scaler.prototype.intercept = function intercept(websocket, req, res, head) {
  if (this.engine && this.endpoint.test(req.url)) {
    if (websocket) return this.engine.handleUpgrade(req, res, head);
    return this.engine.handleRequest(req, res);
  } else if (websocket) {
    //
    // We cannot do fancy pancy redirection for WebSocket calls. So instead we
    // should just close the connecting socket.
    //
    return res.end();
  }

  if ('put' === req.method && this.broadcast.test(req.url)) {
    return this.incoming(req, res);
  }

  //
  // This is an unknown request, let's just assume that this user is just
  // exploring our server with the best intentions and redirect him to the
  // service domain.
  //
  if (this.service) {
    res.statusCode = 301;
    res.setHeader('Location', this.service);
    return res.end('');
  }

  //
  // Well, fuck it, keeel it, with fire!
  //
  this.end('bad request', res);
};

/**
 * Set the network address where this server is accessible on.
 *
 * @param {String} address Network address
 * @param {String} port The port number
 * @api public
 */
Scaler.prototype.network = function network(address, port) {
  if (address) this.address = address;
  if (port) this.port = port;

  return this;
};

/**
 * Add a new connection.
 *
 * @param {String} account Account id.
 * @param {String} session Session id.
 * @param {String} id Connection id
 * @api private
 */
Scaler.prototype.connect = function connect(account, session, id) {
  var key = this.namespace +'::'+ account +'::'+ session
    , value = this.interface +'@'+ id
    , scaler = this;

  this.redis.setx(key, this.timeout, value, function setx(err) {
    if (err) return scaler.emit('error::connect', key, value);
  });

  return this;
};

/**
 * Remove a connection.
 *
 * @param {String} account Account id.
 * @param {String} session Session id.
 * @param {String} id Connection id
 */
Scaler.prototype.disconnect = function disconnect(account, session, id) {
  var key = this.namespace +'::'+ account +'::'+ session
    , value = this.interface +'@'+ id
    , scaler = this;

  this.redis.del(key, function del(err) {
    if (err) return scaler.emit('error::disconnect', key, value);
  });

  return this;
};

/**
 * An other server wants to send something one of our connected sockets.
 *
 * @param {Request} req
 * @param {Respone} res
 * @api private
 */<commit_msg>[minor] Added the .find and .broadcast<commit_after>'use strict';

var Engine = require('engine.io').Server
  , Socket = require('engine.io').Socket
  , request = require('request')
  , Route = require('routable');

/**
 * Create a new Scaler instance.
 *
 * @constructor
 * @param {Redis} redis A Redis client.
 * @param {Object} options Scaler options.
 * @api public
 */
function Scaler(redis, options) {
  options = options || {};

  //
  // The HTTP routes that we should be listening on.
  //
  this.broadcast = new Route(options.broadcast || '/scaler/broadcast');
  this.endpoint = new Route(options.endpoint || '/stream/');

  // The redis client we need to keep connection state.
  this.redis = redis || require('redis').createClient();

  // The root domain of the service, will be used for redirects.
  this.service = options.service || false;

  // The namespace for the keys that are stored in redis.
  this.namespace = options.namespace || 'scaler';

  // How long do we maintain stake from a single user.
  this.timeout = options.timeout || 60 * 15;

  // The network address this server is approachable on for internal HTTP requests.
  this.address = options.address || 'localhost';

  // The port number that we should use to connect with this server.
  this.port = options.port || null;

  //
  // Allow custom encoders and decoders, we default to JSON.parse so it should
  // have simular interfaces.
  //
  this.encode = options.encode || JSON.stringify;
  this.decode = options.decode || JSON.parse;

  //
  // These properies will be set once we're initialized.
  //
  this.server = null;         // HTTP server instance.
  this.engine = null;         // Engine.IO server.
}

//
// The scaler inherits from the EventEmitter so we can savely emit events
// without creating tail recursion.
//
Scaler.prototype.__proto__ = require('events').EventEmitter.prototype;

//
// Because we are to lazy to combine address + port every single time.
//
Object.defineProperty(Scaler.prototype, 'interface', {
  get: function get() {
    return this.address +':'+ this.port;
  }
});

/**
 * Intercept HTTP requests and handle them accordingly.
 *
 * @param {Boolean} websocket This is a WebSocket request
 * @param {Request} req HTTP request instance.
 * @param {Response} res HTTP response instance.
 * @param {Buffer} head HTTP head buffer.
 * @api private
 */
Scaler.prototype.intercept = function intercept(websocket, req, res, head) {
  if (this.engine && this.endpoint.test(req.url)) {
    if (websocket) return this.engine.handleUpgrade(req, res, head);
    return this.engine.handleRequest(req, res);
  } else if (websocket) {
    //
    // We cannot do fancy pancy redirection for WebSocket calls. So instead we
    // should just close the connecting socket.
    //
    return res.end();
  }

  if ('put' === req.method && this.broadcast.test(req.url)) {
    return this.incoming(req, res);
  }

  //
  // This is an unknown request, let's just assume that this user is just
  // exploring our server with the best intentions and redirect him to the
  // service domain.
  //
  if (this.service) {
    res.statusCode = 301;
    res.setHeader('Location', this.service);
    return res.end('');
  }

  //
  // Well, fuck it, keeel it, with fire!
  //
  this.end('bad request', res);
};

/**
 * Set the network address where this server is accessible on.
 *
 * @param {String} address Network address
 * @param {String} port The port number
 * @api public
 */
Scaler.prototype.network = function network(address, port) {
  if (address) this.address = address;
  if (port) this.port = port;

  return this;
};

/**
 * Add a new connection.
 *
 * @param {String} account Account id.
 * @param {String} session Session id.
 * @param {String} id Connection id
 * @api private
 */
Scaler.prototype.connect = function connect(account, session, id) {
  var key = this.namespace +'::'+ account +'::'+ session
    , value = this.interface +'@'+ id
    , scaler = this;

  this.redis.setx(key, this.timeout, value, function setx(err) {
    if (err) return scaler.emit('error::connect', key, value);
  });

  return this;
};

/**
 * Remove a connection.
 *
 * @param {String} account Account id.
 * @param {String} session Session id.
 * @param {String} id Connection id
 */
Scaler.prototype.disconnect = function disconnect(account, session, id) {
  var key = this.namespace +'::'+ account +'::'+ session
    , value = this.interface +'@'+ id
    , scaler = this;

  this.redis.del(key, function del(err) {
    if (err) return scaler.emit('error::disconnect', key, value);
  });

  return this;
};

/**
 * Find a server for a given session id.
 *
 * @param {String} account Observe.it account id.
 * @param {String} session Session id of the user.
 * @param {Function} fn Callback
 * @api private
 */
Scaler.prototype.find = function find(account, session, fn) {
  var key = this.namespace +'::'+ account +'::'+ session;

  this.redis.get(key, function parse(err, data) {
    if (err) return fn(err);

    //
    // The format of the response is <serverhost:port>@<socket id>, just split it.
    //
    data = data.split('@');
    fn(undefined, data[0], data[1]);
  });

  return this;
};

/**
 * Send a message to the given id.
 *
 * @param {String} account Observe.it account id.
 * @param {String} session Session id of the user.
 * @param {String} message Message
 * @param {Function} fn Callback
 * @api public
 */
Scaler.prototype.broadcast = function broadcast(account, session, message, fn) {
  this.find(account, session, function found(err, server, id) {
    if (err || !server) return fn(err || new Error('Unknown session id '+ session));

    request({
      uri: 'http://'+ server + exports.endpoint,
      method: 'PUT',
      json: {
        id: id,           // The id of the socket that should receive the data.
        message: message  // The actual message.
      }
    }, function requested(err, response, body) {
      var status = response.statusCode;

      if (err || status !== 200) {
        return fn(err || new Error('Invalid status code ('+ status +') returned'));
      }

      //
      // We only have successfully send the message when we received
      // a statusCode 200 from the targetted server.
      //
      fn(undefined, body);
    });
  });

  return this;
};

/**
 * An other server wants to send something one of our connected sockets.
 *
 * @param {Request} req
 * @param {Respone} res
 * @api private
 */
{'fraction_non_alphanumeric': 0.1368148081910042, 'fraction_numerical': 0.0019672717517660735, 'mean_word_length': 3.068388504910877, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': 2, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '689307', 'n_tokens_mistral': 3400, 'n_tokens_neox': 3313, 'n_words': 1167}