<commit_before>var util = require('util');
var duplex = require('readable-stream').Duplex;
var split = require('split2');
var ircmessage = require('./irc-message');

var ircstream = module.exports = function(stream) {
    if (!(this instanceof ircstream)) {
        return new ircstream(stream);
    }

    duplex.call(this, {objectMode: true});

    this._readable = stream.pipe(split());
    this._writable = stream;
    this._readRequested = false;

    var self = this;

    this._writable.once('finish', function() {
        self.end();
    });

    this.once('finish', function() {
        self._writable.end();
    });

    self._readable.on('readable', function() {
        self._forwardRead();
    });

    self._readable.once('end', function() {
        return self.push(null);
    });

    self._writable.on('error', function(error) {
      return self.emit('error', error);
    });

    self._readable.on('error', function(error) {
      return self.emit('error', error);
    });
}
util.inherits(ircstream, duplex);

ircstream.prototype._write = function(object, encoding, callback) {
    this._writable.write(ircmessage(object));
    callback();
};

ircstream.prototype._read = function() {
    if (!this._readRequested) {
        this._readRequested = true;
    }
};

ircstream.prototype._forwardRead = function() {
    if (this._readRequested) {
        var data;

        while ((data = this._readable.read()) !== null) {
            this._readRequested = false
            this.push(ircmessage(data));
        }
    }
};<commit_msg>Update _write to pass callback<commit_after>var util = require('util');
var duplex = require('readable-stream').Duplex;
var split = require('split2');
var ircmessage = require('./irc-message');

var ircstream = module.exports = function(stream) {
    if (!(this instanceof ircstream)) {
        return new ircstream(stream);
    }

    duplex.call(this, {objectMode: true});

    this._readable = stream.pipe(split());
    this._writable = stream;
    this._readRequested = false;

    var self = this;

    this._writable.once('finish', function() {
        self.end();
    });

    this.once('finish', function() {
        self._writable.end();
    });

    self._readable.on('readable', function() {
        self._forwardRead();
    });

    self._readable.once('end', function() {
        return self.push(null);
    });

    self._writable.on('error', function(error) {
      return self.emit('error', error);
    });

    self._readable.on('error', function(error) {
      return self.emit('error', error);
    });
}
util.inherits(ircstream, duplex);

ircstream.prototype._write = function(object, encoding, callback) {
    this._writable.write(ircmessage(object), callback);
};

ircstream.prototype._read = function() {
    if (!this._readRequested) {
        this._readRequested = true;
    }
};

ircstream.prototype._forwardRead = function() {
    if (this._readRequested) {
        var data;

        while ((data = this._readable.read()) !== null) {
            this._readRequested = false
            this.push(ircmessage(data));
        }
    }
};
{'fraction_non_alphanumeric': 0.16211000321646832, 'fraction_numerical': 0.000643293663557414, 'mean_word_length': 2.921815889029004, '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': '2677448', 'n_tokens_mistral': 972, 'n_tokens_neox': 919, 'n_words': 174}