<commit_before>var Stream = require('stream');

// writable
exports.Event = function (options) {
   
    options = options || {};
    if (typeof options.objectMode === 'undefined') {
        options.objectMode = true;
    }
 
    var sequence = Stream.Duplex({
        readableObjectMode: options.objectMode ? true : false,
        writableObjectMode: options.objectMode ? true : false
    });

    // writable
    sequence._write = function (chunk, enc, next) {
        
        if (!sequence.seq) {
            return sequence.emit('error', new Error('Stream._write: Sequence not set.'));
        }

        sequence._next = next;
        sequence._pos = -1;
        sequence.next(null, chunk);
    };

    // readable
    sequence._read = function () {
        if (sequence.seq) {
            sequence.uncork();
        }
    };


    // next handler
    sequence.next = function (err, data) {

        // handle steam end (null) | (false, null)
        if (err === null && (data === null || data === undefined)) {
            sequence.end();
            return;
        }

        // set error flag and reset position (err)
        if (!sequence._err && err) {
            sequence._err = true;
            sequence._pos = 0;
            data = err;
        }

        // get handler
        // TODO handle no handler (reset position if no handler is found?)
        var handler = sequence.seq[sequence._err ? 'error' : 'data'][++sequence._pos];
        if (!handler) {

            // emit error to reader
            if (sequence._err) {
                sequence._next(data || new Error('Error stream: no data.'));
                return;
            }

            // auto convert to non-object streams
            if (!options.objectMode &&
                typeof data !== 'string' &&
                data.constructor !== Buffer
            ) {
                try {
                    data = JSON.stringify(data);
                } catch (err)  {
                    sequence._next(err);
                    return;
                }
            }

            // write to linked stream
            if (sequence.linked) {

                // push also to response
                if (sequence.seq.type === '|') {
                    sequence.push(data);
                }

                // handle back pressure
                if (!sequence.linked.write(data)) {
                    sequence.linked.once('drain', sequence._next);
                    return;
                }
                
            // push data to readable
            } else if ((!sequence.push(data))) {

                // buffer all writes
                sequence.cork()
            }

            // request the next data chunk
            sequence._next();
            return;
        }

        // call next data handler
        handler[1].call(handler[0], sequence.next.bind(sequence), handler[2], data);
    };

    return sequence;
};
<commit_msg>working on data handler calls<commit_after>var Stream = require('stream');

// writable
exports.Event = function (options) {
   
    options = options || {};
    if (typeof options.objectMode === 'undefined') {
        options.objectMode = true;
    }
 
    var sequence = Stream.Duplex({
        readableObjectMode: options.objectMode ? true : false,
        writableObjectMode: options.objectMode ? true : false
    });

    // writable
    sequence._write = function (chunk, enc, next) {

        if (!sequence.seq) {
            return sequence.emit('error', new Error('Stream._write: Sequence not set.'));
        }

        sequence._next = next;
        sequence._pos = -1;
        sequence.next(null, chunk);
    };

    // readable
    sequence._read = function () {
        if (sequence.seq) {
            sequence.uncork();
        }
    };


    // next haninstancedler
    sequence.next = function (err, data) {

        // handle steam end (null) | (false, null)
        if (err === null && (data === null || data === undefined)) {
            sequence.end();
            return;
        }

        // set error flag and reset position (err)
        if (!sequence._err && err) {
            sequence._err = true;
            sequence._pos = 0;
            data = err;
        }

        // get handler
        // TODO handle no handler (reset position if no handler is found?)
        var handler = sequence.seq[0][++sequence._pos];
        if (!handler) {

            // emit error to reader
            if (sequence._err) {
                sequence._next(data || new Error('Error stream: no data.'));
                return;
            }

            // auto convert to non-object streams
            if (!options.objectMode &&
                typeof data !== 'string' &&
                data.constructor !== Buffer
            ) {
                try {
                    data = JSON.stringify(data);
                } catch (err)  {
                    sequence._next(err);
                    return;
                }
            }

            // write to linked stream
            if (sequence.linked) {

                // push also to response
                if (sequence.seq[1][0] === '|') {
                    sequence.push(data);
                }

                // handle back pressure
                if (!sequence.linked.write(data)) {
                    sequence.linked.once('drain', sequence._next);
                    return;
                }
                
            // push data to readable
            } else if ((!sequence.push(data))) {

                // buffer all writes
                sequence.cork()
            }

            // request the next data chunk
            sequence._next();
            return;
        }

        // call next data handler
        handler[0].call(handler[2], sequence.next.bind(sequence), handler[1], data);
    };

    return sequence;
};

{'fraction_non_alphanumeric': 0.11955040871934605, 'fraction_numerical': 0.0022138964577656677, 'mean_word_length': 1.2896686159844055, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '612053', 'n_tokens_mistral': 1507, 'n_tokens_neox': 1416, 'n_words': 384}