<commit_before>/*
 * node-qtdatastream
 * https://github.com/magne4000/node-qtdatastream
 *
 * Copyright (c) 2016 <NAME>
 * Licensed under the MIT license.
 */

var EventEmitter = require('events').EventEmitter,
    NetSocket = require('net').Socket,
    debuglib = require('debug'),
    logger = debuglib('qtdatastream:main'),
    loggerw = debuglib('qtdatastream:main-write'),
    util = require('util'),
    debug = !!process.env.QTDSDEBUG || debuglib.enabled("qtdatastream:*");
    
/** @module qtdatastream */

/**
 * Qt types list
 * @readonly
 * @enum {number}
 */
exports.Types = {
    INVALID: 0,
    BOOL: 1,
    INT: 2,
    UINT: 3,
    INT64: 4,
    UINT64: 5,
    DOUBLE: 6,
    CHAR: 7,
    MAP: 8,
    LIST: 9,
    STRING: 10,
    STRINGLIST: 11,
    BYTEARRAY: 12,
    TIME: 15,
    DATETIME: 16,
    USERTYPE: 127,
    SHORT: 133
};

if (debug && !debuglib.enabled("qtdatastream:*")) {
    debuglib.enable("qtdatastream:*");
}

exports.QTDATASTREAMCLASS = "__QTDATASTREAMCLASS__";

exports.userTypes = {};

exports.toggleEndianness = function(buffer) {
    buffer.swap16();
    return buffer;
};

/**
 * Qt compliant Socket overload.
 * 'data' event is triggered only when full buffer is received.
 * 'error', 'close' and 'end' event are not altered.
 * @class
 * @param {net.Socket} socket
 */
exports.Socket = function Socket(socket, readCallback, writeCallback) {
    if (!(socket instanceof NetSocket)) {
        throw "Socket must be an instance of net.Socket";
    }
    var self = this;
    self.socket = socket;
    self.data_state = null;
    self.readCallback = readCallback;
    self.writeCallback = writeCallback;
    
    self.updateSocket(socket, readCallback, writeCallback);
};
util.inherits(exports.Socket, EventEmitter);

/**
 * Write data to the socket
 * @param {*} data Data that will be written using Writer
 */
exports.Socket.prototype.write = function(data) {
    var Writer = require('./writer');
    var writer = new Writer(data);
    var buffer = writer.getBuffer();
    var self = this;
    if (typeof this.writeCallback === 'function') {
        this.writeCallback(buffer, function(err, buffer2) {
            if (!err) {
                if (debug) {
                    loggerw(buffer2);
                }
                self.socket.write(buffer2);
            } else {
                logger(err);
            }
        });
    } else {
        if (debug) {
            loggerw(buffer);
        }
        this.socket.write(buffer);
    }
};

/**
 * Update the socket (for example to promote it to SSL stream)
 * @param {Stream} socket object implementing Stream interface
 */
exports.Socket.prototype.updateSocket = function(socket) {
    var Reader = require('./reader');
    var self = this;

    this.socket.removeAllListeners();

    this.socket = socket;

    this.socket.on('data', function(data) {
        if (typeof self.readCallback === 'function') {
            self.readCallback(data, handleData);
        } else {
            handleData(null, data);
        }
    });

    function getsize(bufferlist) {
        // get 4 bytes
        var final_buffer;
        if (bufferlist[0] && bufferlist[0].length >= 4) {
            final_buffer = bufferlist[0];
        } else {
            var totallength = 0, i = 0;
            while(totallength < 4 && i < bufferlist.length) {
                totallength += bufferlist[i++].length;
            }
            if (totallength < 4) return Infinity;
            final_buffer = Buffer.concat(bufferlist.slice(0,i), totallength);
        }
        var reader = new Reader(final_buffer);
        return reader.size;
    }

    function handleData(err, data) {
        var reader;
        if (!err) {
            var stop = false;
            while (!stop) {
                if (!self.data_state) {
                    self.data_state = {size: Infinity, data: [], recvd: 0};
                }
                var ds = self.data_state;
                if (data !== null) {
                    ds.data.push(data);
                    ds.recvd += data.length;
                }
                if (ds.size == Infinity && ds.recvd >= 4) {
                    ds.size = getsize(ds.data);
                }
                self.emit('progress', ds.recvd, ds.size);
                if (ds.size > ds.recvd) {
                    if (debug) {
                        logger("(%d/%d) Waiting for end of buffer", ds.recvd, ds.size);
                    }
                    stop = true;
                } else {
                    reader = new Reader(Buffer.concat(ds.data, ds.recvd));
                    if (debug) {
                        logger("(%d/%d) Received full buffer", ds.recvd, ds.size);
                    }
                    reader.parse();
                    if (debug) {
                        logger('Received result');
                        logger(reader.parsed);
                    }
                    if (reader.remaining && reader.remaining.length > 0) {
                        self.data_state = {
                            data: [reader.remaining],
                            recvd: reader.remaining.length,
                            size: Infinity
                        };
                        stop = false;
                    } else {
                        self.data_state = null;
                        stop = true;
                    }
                    self.emit('data', reader.parsed);
                }
                if (!stop) {
                    data = null;
                }
            }<commit_msg>Fix off-by-4 error in handleData<commit_after>/*
 * node-qtdatastream
 * https://github.com/magne4000/node-qtdatastream
 *
 * Copyright (c) 2016 <NAME>
 * Licensed under the MIT license.
 */

var EventEmitter = require('events').EventEmitter,
    NetSocket = require('net').Socket,
    debuglib = require('debug'),
    logger = debuglib('qtdatastream:main'),
    loggerw = debuglib('qtdatastream:main-write'),
    util = require('util'),
    debug = !!process.env.QTDSDEBUG || debuglib.enabled("qtdatastream:*");
    
/** @module qtdatastream */

/**
 * Qt types list
 * @readonly
 * @enum {number}
 */
exports.Types = {
    INVALID: 0,
    BOOL: 1,
    INT: 2,
    UINT: 3,
    INT64: 4,
    UINT64: 5,
    DOUBLE: 6,
    CHAR: 7,
    MAP: 8,
    LIST: 9,
    STRING: 10,
    STRINGLIST: 11,
    BYTEARRAY: 12,
    TIME: 15,
    DATETIME: 16,
    USERTYPE: 127,
    SHORT: 133
};

if (debug && !debuglib.enabled("qtdatastream:*")) {
    debuglib.enable("qtdatastream:*");
}

exports.QTDATASTREAMCLASS = "__QTDATASTREAMCLASS__";

exports.userTypes = {};

exports.toggleEndianness = function(buffer) {
    buffer.swap16();
    return buffer;
};

/**
 * Qt compliant Socket overload.
 * 'data' event is triggered only when full buffer is received.
 * 'error', 'close' and 'end' event are not altered.
 * @class
 * @param {net.Socket} socket
 */
exports.Socket = function Socket(socket, readCallback, writeCallback) {
    if (!(socket instanceof NetSocket)) {
        throw "Socket must be an instance of net.Socket";
    }
    var self = this;
    self.socket = socket;
    self.data_state = null;
    self.readCallback = readCallback;
    self.writeCallback = writeCallback;
    
    self.updateSocket(socket, readCallback, writeCallback);
};
util.inherits(exports.Socket, EventEmitter);

/**
 * Write data to the socket
 * @param {*} data Data that will be written using Writer
 */
exports.Socket.prototype.write = function(data) {
    var Writer = require('./writer');
    var writer = new Writer(data);
    var buffer = writer.getBuffer();
    var self = this;
    if (typeof this.writeCallback === 'function') {
        this.writeCallback(buffer, function(err, buffer2) {
            if (!err) {
                if (debug) {
                    loggerw(buffer2);
                }
                self.socket.write(buffer2);
            } else {
                logger(err);
            }
        });
    } else {
        if (debug) {
            loggerw(buffer);
        }
        this.socket.write(buffer);
    }
};

/**
 * Update the socket (for example to promote it to SSL stream)
 * @param {Stream} socket object implementing Stream interface
 */
exports.Socket.prototype.updateSocket = function(socket) {
    var Reader = require('./reader');
    var self = this;

    this.socket.removeAllListeners();

    this.socket = socket;

    this.socket.on('data', function(data) {
        if (typeof self.readCallback === 'function') {
            self.readCallback(data, handleData);
        } else {
            handleData(null, data);
        }
    });

    function getsize(bufferlist) {
        // get 4 bytes
        var final_buffer;
        if (bufferlist[0] && bufferlist[0].length >= 4) {
            final_buffer = bufferlist[0];
        } else {
            var totallength = 0, i = 0;
            while(totallength < 4 && i < bufferlist.length) {
                totallength += bufferlist[i++].length;
            }
            if (totallength < 4) return Infinity;
            final_buffer = Buffer.concat(bufferlist.slice(0,i), totallength);
        }
        var reader = new Reader(final_buffer);
        return reader.size;
    }

    function handleData(err, data) {
        var reader;
        if (!err) {
            var stop = false;
            while (!stop) {
                if (!self.data_state) {
                    self.data_state = {size: Infinity, data: [], recvd: 0};
                }
                var ds = self.data_state;
                if (data !== null) {
                    ds.data.push(data);
                    ds.recvd += data.length;
                }
                if (ds.size == Infinity && ds.recvd >= 4) {
                    ds.size = getsize(ds.data);
                }
                self.emit('progress', ds.recvd, ds.size);
                if (ds.size + 4 > ds.recvd) {
                    if (debug) {
                        logger("(%d/%d) Waiting for end of buffer", ds.recvd, ds.size + 4);
                    }
                    stop = true;
                } else {
                    reader = new Reader(Buffer.concat(ds.data, ds.recvd));
                    if (debug) {
                        logger("(%d/%d) Received full buffer", ds.recvd, ds.size + 4);
                    }
                    reader.parse();
                    if (debug) {
                        logger('Received result');
                        logger(reader.parsed);
                    }
                    if (reader.remaining && reader.remaining.length > 0) {
                        self.data_state = {
                            data: [reader.remaining],
                            recvd: reader.remaining.length,
                            size: Infinity
                        };
                        stop = false;
                    } else {
                        self.data_state = null;
                        stop = true;
                    }
                    self.emit('data', reader.parsed);
                }
                if (!stop) {
                    data = null;
                }
            }
{'fraction_non_alphanumeric': 0.12447276316970295, 'fraction_numerical': 0.010410122947141703, 'mean_word_length': 1.5472, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 2, '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': '5659734', 'n_tokens_mistral': 3299, 'n_tokens_neox': 3080, 'n_words': 741}