<commit_before>'use strict';

var _ = require('lodash');
var Command = require('./command');
var Commander = require('./commander');
var Promise = require('bluebird');
var utils = require('./utils');

function Pipeline(redis) {
  Commander.call(this);

  this.redis = redis;
  this._queue = [];
  this._result = [];
  this._transactions = 0;
  this._shaToScript = {};

  var _this = this;
  _.keys(redis.scriptsSet).forEach(function (name) {
    var script = redis.scriptsSet[name];
    _this._shaToScript[script.sha] = script;
    _this[name] = redis[name];
    _this[name + 'Buffer'] = redis[name + 'Buffer'];
  });

  this.promise = new Promise(function (resolve, reject) {
    _this.resolve = resolve;
    _this.reject = reject;
  });
}

_.extend(Pipeline.prototype, Commander.prototype);

Pipeline.prototype.fillResult = function (value, position) {
  this._result[position] = value;
  if (!--this.replyPending) {
    this.resolve(this._result);
  }
};

Pipeline.prototype.sendCommand = function (command) {
  var position = this._queue.length;

  var _this = this;
  command.promise.then(function (result) {
    _this.fillResult([null, result], position);
  }).catch(function (error) {
    _this.fillResult([error], position);
  });

  this._queue.push(command);

  return this;
};

var multi = Pipeline.prototype.multi;
Pipeline.prototype.multi = function () {
  this._transactions += 1;
  return multi.apply(this, arguments);
};

var exec = Pipeline.prototype.exec;
Pipeline.prototype.exec = function (callback) {
  if (this._transactions > 0) {
    this._transactions -= 1;
    return exec.apply(this, arguments);
  }
  var isCluster = this.redis.constructor.name === 'Cluster';

  var sampleKey;
  // Check whether scripts exists and get a sampleKey.
  var scripts = [];
  for (var i = 0; i < this._queue.length; ++i) {
    var item = this._queue[i];
    if (isCluster && !sampleKey) {
      sampleKey = item.getKeys()[0];
    }
    if (item.name !== 'evalsha') {
      continue;
    }
    var script = this._shaToScript[item.args[0]];
    if (!script) {
      continue;
    }
    scripts.push(script);
  }
  var sampleSlot;
  if (isCluster) {
    if (sampleKey) {
      sampleSlot = utils.calcSlot(sampleKey);
    } else {
      sampleSlot = Math.random() * 16384 | 0;
    }
  }

  var promise;
  var _this = this;
  if (scripts.length) {
    promise = this.redis.script('exists', scripts.map(function (item) {
      return item.sha;
    })).then(function (results) {
      var pending = [];
      for (var i = 0; i < results.length; ++i) {
        if (!results[i]) {
          pending.push(scripts[i]);
        }
      }
      return Promise.all(pending.map(function (script) {
        return _this.redis.script('load', script.lua);
      }));
    });
  } else {
    promise = Promise.resolve();
  }

  return promise.then(function () {
    var data = '';
    var writePending = _this.replyPending =  _this._queue.length;
    // TODO optimize buffer
    var stream = {
      write: function (writable) {
        data += writable;
        if (!--writePending) {
          if (isCluster) {
            _this.redis.slots[sampleSlot].stream.write(data);
          } else {
            _this.redis.stream.write(data);
          }
        }
      }
    };

    for (var i = 0; i < _this._queue.length; ++i) {
      if (_this._queue[i] instanceof Command) {
        _this.redis.sendCommand(_this._queue[i], stream, sampleSlot);
      }<commit_msg>Support write buffer in pipeline. #20<commit_after>'use strict';

var _ = require('lodash');
var Command = require('./command');
var Commander = require('./commander');
var fbuffer = require('flexbuffer');
var Promise = require('bluebird');
var utils = require('./utils');

function Pipeline(redis) {
  Commander.call(this);

  this.redis = redis;
  this._queue = [];
  this._result = [];
  this._transactions = 0;
  this._shaToScript = {};

  var _this = this;
  _.keys(redis.scriptsSet).forEach(function (name) {
    var script = redis.scriptsSet[name];
    _this._shaToScript[script.sha] = script;
    _this[name] = redis[name];
    _this[name + 'Buffer'] = redis[name + 'Buffer'];
  });

  this.promise = new Promise(function (resolve, reject) {
    _this.resolve = resolve;
    _this.reject = reject;
  });
}

_.extend(Pipeline.prototype, Commander.prototype);

Pipeline.prototype.fillResult = function (value, position) {
  this._result[position] = value;
  if (!--this.replyPending) {
    this.resolve(this._result);
  }
};

Pipeline.prototype.sendCommand = function (command) {
  var position = this._queue.length;

  var _this = this;
  command.promise.then(function (result) {
    _this.fillResult([null, result], position);
  }).catch(function (error) {
    _this.fillResult([error], position);
  });

  this._queue.push(command);

  return this;
};

var multi = Pipeline.prototype.multi;
Pipeline.prototype.multi = function () {
  this._transactions += 1;
  return multi.apply(this, arguments);
};

var exec = Pipeline.prototype.exec;
Pipeline.prototype.exec = function (callback) {
  if (this._transactions > 0) {
    this._transactions -= 1;
    return exec.apply(this, arguments);
  }
  var isCluster = this.redis.constructor.name === 'Cluster';

  var sampleKey;
  // Check whether scripts exists and get a sampleKey.
  var scripts = [];
  for (var i = 0; i < this._queue.length; ++i) {
    var item = this._queue[i];
    if (isCluster && !sampleKey) {
      sampleKey = item.getKeys()[0];
    }
    if (item.name !== 'evalsha') {
      continue;
    }
    var script = this._shaToScript[item.args[0]];
    if (!script) {
      continue;
    }
    scripts.push(script);
  }
  var sampleSlot;
  if (isCluster) {
    if (sampleKey) {
      sampleSlot = utils.calcSlot(sampleKey);
    } else {
      sampleSlot = Math.random() * 16384 | 0;
    }
  }

  var promise;
  var _this = this;
  if (scripts.length) {
    promise = this.redis.script('exists', scripts.map(function (item) {
      return item.sha;
    })).then(function (results) {
      var pending = [];
      for (var i = 0; i < results.length; ++i) {
        if (!results[i]) {
          pending.push(scripts[i]);
        }
      }
      return Promise.all(pending.map(function (script) {
        return _this.redis.script('load', script.lua);
      }));
    });
  } else {
    promise = Promise.resolve();
  }

  return promise.then(function () {
    var data = '';
    var writePending = _this.replyPending =  _this._queue.length;

    var bufferMode = false;
    var stream = {
      write: function (writable) {
        if (writable instanceof Buffer) {
          bufferMode = true;
        }
        if (bufferMode) {
          if (typeof data === 'string') {
            var flexBuffer = new fbuffer.FlexBuffer();
            flexBuffer.write(data);
            data = flexBuffer;
          }
          data.write(writable);
        } else {
          data += writable;
        }
        if (!--writePending) {
          if (bufferMode) {
            data = data.getBuffer();
          }
          if (isCluster) {
            _this.redis.slots[sampleSlot].stream.write(data);
          } else {
            _this.redis.stream.write(data);
          }
        }
      }
    };

    for (var i = 0; i < _this._queue.length; ++i) {
      if (_this._queue[i] instanceof Command) {
        _this.redis.sendCommand(_this._queue[i], stream, sampleSlot);
      }
{'fraction_non_alphanumeric': 0.1520238256396372, 'fraction_numerical': 0.004331934479490998, 'mean_word_length': 2.632251720747296, 'pattern_counts': {'":': 0, '<': 9, '<?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': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '297471', 'n_tokens_mistral': 2386, 'n_tokens_neox': 2354, 'n_words': 507}