<commit_before>var net = require('net');
var Stream = require('stream');
var protobuf = require('protobuf.js');
var butils = require('butils');
var through = require('through');
var path = require('path');
var _merge = require('./lib/merge');
var parseContent = require('./lib/parse-content');

var messageCodes = {
    '0': 'RpbErrorResp',
    '1': 'RpbPingReq',
    '2': 'RpbPingResp',
    '3': 'RpbGetClientIdReq',
    '4': 'RpbGetClientIdResp',
    '5': 'RpbSetClientIdReq',
    '6': 'RpbSetClientIdResp',
    '7': 'RpbGetServerInfoReq',
    '8': 'RpbGetServerInfoResp',
    '9': 'RpbGetReq',
    '10': 'RpbGetResp',
    '11': 'RpbPutReq',
    '12': 'RpbPutResp',
    '13': 'RpbDelReq',
    '14': 'RpbDelResp',
    '15': 'RpbListBucketsReq',
    '16': 'RpbListBucketsResp',
    '17': 'RpbListKeysReq',
    '18': 'RpbListKeysResp',
    '19': 'RpbGetBucketReq',
    '20': 'RpbGetBucketResp',
    '21': 'RpbSetBucketReq',
    '22': 'RpbSetBucketResp',
    '23': 'RpbMapRedReq',
    '24': 'RpbMapRedResp',
    '25': 'RpbIndexReq',
    '26': 'RpbIndexResp',
    '27': 'RpbSearchQueryReq',
    '28': 'RpbSearchQueryResp',
    // 1.4
    '29': 'RpbResetBucketReq',
    '30': 'RpbResetBucketResp',
    '40': 'RpbCSBucketReq',
    '41': 'RpbCSBucketResp',
    '50': 'RpbCounterUpdateReq',
    '51': 'RpbCounterUpdateResp',
    '52': 'RpbCounterGetReq',
    '53': 'RpbCounterGetResp',
};

Object.keys(messageCodes).forEach(function (key) {
    messageCodes[messageCodes[key]] = Number(key);
});

function RiakPBC(options) {
    var self = this;
    options = options || {};
    self.host = options.host || '127.0.0.1';
    self.port = options.port || 8087;
    self.timeout = options.timeout || 1000;
    self.bucket = options.bucket || undefined;
    self.translator = protobuf.loadSchema(path.join(__dirname, './spec/riak_kv.proto'));
    self.client = new net.Socket();
    self.connected = false;
    self.client.on('end', self.disconnect.bind(this));
    self.client.on('error', self.disconnect.bind(this));
    self.client.on('data', self._processPacket.bind(this));
    self.paused = false;
    self.queue = [];
    self.reply = {};
    self.resBuffers = [];
    self.numBytesAwaiting = 0;
}

RiakPBC.prototype._splitPacket = function (pkt) {
    var self = this;
    var pos = 0;
    var len;

    if (self.numBytesAwaiting > 0) {
        len = Math.min(pkt.length, self.numBytesAwaiting);
        var oldBuf = self.resBuffers[self.resBuffers.length - 1];
        var newBuf = new Buffer(oldBuf.length + len);
        oldBuf.copy(newBuf, 0);
        pkt.slice(0, len).copy(newBuf, oldBuf.length);
        self.resBuffers[self.resBuffers.length - 1] = newBuf;
        pos = len;
        self.numBytesAwaiting -= len;
    } else {
        self.resBuffers = [];
    }

    while (pos < pkt.length) {
        len = butils.readInt32(pkt, pos);
        self.numBytesAwaiting = len + 4 - pkt.length;
        self.resBuffers.push(pkt.slice(pos + 4, Math.min(pos + len + 4, pkt.length)));
        pos += len + 4;
    }
};

RiakPBC.prototype._processPacket = function (chunk) {
    var self = this;

    self._splitPacket(chunk);
    if (self.numBytesAwaiting > 0) {
        return;
    }
    processAllResBuffers.call(self, self.resBuffers);
};

function processAllResBuffers(resBuffers) {
    var self = this;
    var stream = self.task.stream;
    var mc, err, cb;

    resBuffers.forEach(processSingleResBuffer);

    if (!self.task.expectMultiple || self.reply.done || mc === 'RpbErrorResp') {
        cb = self.task.callback;
        self.task = undefined;

        if (stream) {
            stream.end();
        } else {
            cb(err, self.reply);
        }

        mc = undefined;
        self.reply = {};
        self.paused = false;
        self._processNext();
    }

    function processSingleResBuffer(packet) {
        var response;

        mc = messageCodes['' + packet[0]];
        response = self.translator.decode(mc, packet.slice(1));
        if (response.content && Array.isArray(response.content)) {
            response.content.forEach(parseContent);
        }

        if (response.errmsg) {
            err = new Error(response.errmsg);
            err.code = response.errcode;
            if (stream) {
                stream.emit('error', err);
                return;
            }
        }

        if (stream && !response.done) {
            stream.write(response);
        }

        if (stream) {
            self.reply = response;
        } else {
            self.reply = _merge(self.reply, response);
        }
    }
}

RiakPBC.prototype._processNext = function () {
    var self = this;

    if (self.queue.length && !self.paused) {
        self.paused = true;
        self.connect(function (err) {
            self.task = self.queue.shift();

            if (err) {
                if (self.task.callback) {
                    self.task.callback(err);
                    return;
                }
                if (self.task.stream) {
                    self.task.stream.emit('error', err);
                    return;
                }

                // unhandled error, throw it
                throw err;

            }

            self.client.write(self.task.message);
        });<commit_msg>minor cleanup, attempt to fix bug where self.task ended up as undefined<commit_after>var net = require('net');
var Stream = require('stream');
var protobuf = require('protobuf.js');
var butils = require('butils');
var through = require('through');
var path = require('path');
var _merge = require('./lib/merge');
var parseContent = require('./lib/parse-content');

var messageCodes = {
    '0': 'RpbErrorResp',
    '1': 'RpbPingReq',
    '2': 'RpbPingResp',
    '3': 'RpbGetClientIdReq',
    '4': 'RpbGetClientIdResp',
    '5': 'RpbSetClientIdReq',
    '6': 'RpbSetClientIdResp',
    '7': 'RpbGetServerInfoReq',
    '8': 'RpbGetServerInfoResp',
    '9': 'RpbGetReq',
    '10': 'RpbGetResp',
    '11': 'RpbPutReq',
    '12': 'RpbPutResp',
    '13': 'RpbDelReq',
    '14': 'RpbDelResp',
    '15': 'RpbListBucketsReq',
    '16': 'RpbListBucketsResp',
    '17': 'RpbListKeysReq',
    '18': 'RpbListKeysResp',
    '19': 'RpbGetBucketReq',
    '20': 'RpbGetBucketResp',
    '21': 'RpbSetBucketReq',
    '22': 'RpbSetBucketResp',
    '23': 'RpbMapRedReq',
    '24': 'RpbMapRedResp',
    '25': 'RpbIndexReq',
    '26': 'RpbIndexResp',
    '27': 'RpbSearchQueryReq',
    '28': 'RpbSearchQueryResp',
    // 1.4
    '29': 'RpbResetBucketReq',
    '30': 'RpbResetBucketResp',
    '40': 'RpbCSBucketReq',
    '41': 'RpbCSBucketResp',
    '50': 'RpbCounterUpdateReq',
    '51': 'RpbCounterUpdateResp',
    '52': 'RpbCounterGetReq',
    '53': 'RpbCounterGetResp',
};

Object.keys(messageCodes).forEach(function (key) {
    messageCodes[messageCodes[key]] = Number(key);
});

function RiakPBC(options) {
    var self = this;
    options = options || {};
    self.host = options.host || '127.0.0.1';
    self.port = options.port || 8087;
    self.timeout = options.timeout || 1000;
    self.bucket = options.bucket || undefined;
    self.translator = protobuf.loadSchema(path.join(__dirname, './spec/riak_kv.proto'));
    self.client = new net.Socket();
    self.connected = false;
    self.client.on('end', self.disconnect.bind(this));
    self.client.on('error', self.disconnect.bind(this));
    self.client.on('data', self._processPacket.bind(this));
    self.paused = false;
    self.queue = [];
    self.reply = {};
    self.resBuffers = [];
    self.numBytesAwaiting = 0;
}

RiakPBC.prototype._splitPacket = function (pkt) {
    var self = this;
    var pos = 0;
    var len;

    if (self.numBytesAwaiting > 0) {
        len = Math.min(pkt.length, self.numBytesAwaiting);
        var oldBuf = self.resBuffers[self.resBuffers.length - 1];
        var newBuf = new Buffer(oldBuf.length + len);
        oldBuf.copy(newBuf, 0);
        pkt.slice(0, len).copy(newBuf, oldBuf.length);
        self.resBuffers[self.resBuffers.length - 1] = newBuf;
        pos = len;
        self.numBytesAwaiting -= len;
    } else {
        self.resBuffers = [];
    }

    while (pos < pkt.length) {
        len = butils.readInt32(pkt, pos);
        self.numBytesAwaiting = len + 4 - pkt.length;
        self.resBuffers.push(pkt.slice(pos + 4, Math.min(pos + len + 4, pkt.length)));
        pos += len + 4;
    }
};

RiakPBC.prototype._processPacket = function (chunk) {
    var self = this;

    self._splitPacket(chunk);
    if (self.numBytesAwaiting > 0) {
        return;
    }
    self._processAllResBuffers();
};

RiakPBC.prototype._processAllResBuffers = function () {
    var self = this;
    var stream = self.task.stream;
    var cb = self.task.callback;
    var mc, err;

    self.resBuffers.forEach(processSingleResBuffer);

    if (!self.task.expectMultiple || self.reply.done || mc === 'RpbErrorResp') {
        self.task = undefined;

        if (stream) {
            stream.end();
        } else {
            cb(err, self.reply);
        }

        mc = undefined;
        self.reply = {};
        self.paused = false;
        self._processNext();
    }

    function processSingleResBuffer(packet) {
        var response;

        mc = messageCodes['' + packet[0]];
        response = self.translator.decode(mc, packet.slice(1));
        if (response.content && Array.isArray(response.content)) {
            response.content.forEach(parseContent);
        }

        if (response.errmsg) {
            err = new Error(response.errmsg);
            err.code = response.errcode;
            if (stream) {
                stream.emit('error', err);
                return;
            }
        }

        if (stream && !response.done) {
            stream.write(response);
        }

        if (stream) {
            self.reply = response;
        } else {
            self.reply = _merge(self.reply, response);
        }
    }
};

RiakPBC.prototype._processNext = function () {
    var self = this;

    if (self.queue.length && !self.paused) {
        self.paused = true;
        self.connect(function (err) {
            self.task = self.queue.shift();

            if (!self.task) {
                return;
            }

            if (err) {
                if (self.task.callback) {
                    self.task.callback(err);
                    return;
                }
                if (self.task.stream) {
                    self.task.stream.emit('error', err);
                    return;
                }

                // unhandled error, throw it
                throw err;

            }

            self.client.write(self.task.message);
        });
{'fraction_non_alphanumeric': 0.15492692126355492, 'fraction_numerical': 0.018104667609618104, 'mean_word_length': 2.203261854424645, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '150530', 'n_tokens_mistral': 3674, 'n_tokens_neox': 3482, 'n_words': 655}