<commit_before>var Duplex = require('stream').Duplex;

module.exports = function() {
  var stream = new Duplex();
  var pendingWrite = null;

  var messageListener = function(event) {
    var data = event.data;

    if (data instanceof ArrayBuffer) {
      data = new Buffer(new Uint8Array(data));
    }

    stream.push(data);
  };

  var socketOpen = function() {
    return stream.socket && stream.socket.readyState === 1;
  };

  var openListener = function() {
    if (pendingWrite) {
      socketSend(pendingWrite.chunk, pendingWrite.callback);
      pendingWrite = null;
    }
  };

  var socketSend = function(chunk, callback) {
    if (stream.socket.send.length === 1) {
      try {
        stream.socket.send(chunk);
      } catch (err) {
        return callback(err);
      }

      callback();
      return;
    }

    stream.socket.send(chunk, callback);
  };

  stream.socket = null;

  stream.attach = function(socket) {
    if (this.socket) {
      if ('removeEventListener' in this.socket) {
        this.socket.removeEventListener('message', messageListener);
        this.socket.removeEventListener('open', openListener);
      } else {
        this.socket.removeListener('message', messageListener);
        this.socket.removeListener('open', openListener);
      }
    }

    this.socket = socket;
    this.socket.addEventListener('message', messageListener);
    this.socket.addEventListener('open', openListener);

    if (this.socket.binaryType === 'blob') {
      this.socket.binaryType = 'arraybuffer';
    }

    return this;
  };

  stream._write = function(chunk, encoding, callback) {
    if (!socketOpen()) {
      pendingWrite = { chunk: chunk, callback: callback };
      return;
    }

    socketSend(chunk, callback);
  };

  stream._read = function(size) {};

  return stream;
};<commit_msg>Minor refactor.<commit_after>var Duplex = require('stream').Duplex;

module.exports = function() {
  var stream = new Duplex();
  var pendingWrite = null;

  var messageListener = function(event) {
    var data = event.data;

    if (data instanceof ArrayBuffer) {
      data = new Buffer(new Uint8Array(data));
    }

    stream.push(data);
  };

  var isSocketOpen = function() {
    return stream.socket && stream.socket.readyState === 1;
  };

  var openListener = function() {
    if (pendingWrite) {
      socketSend(pendingWrite.chunk, pendingWrite.callback);
      pendingWrite = null;
    }
  };

  var socketSend = function(chunk, callback) {
    if (stream.socket.send.length === 1) {
      try {
        stream.socket.send(chunk);
      } catch (err) {
        return callback(err);
      }

      callback();
      return;
    }

    stream.socket.send(chunk, callback);
  };

  stream.socket = null;

  stream.attach = function(socket) {
    if (this.socket) {
      if ('removeEventListener' in this.socket) {
        this.socket.removeEventListener('message', messageListener);
        this.socket.removeEventListener('open', openListener);
      } else {
        this.socket.removeListener('message', messageListener);
        this.socket.removeListener('open', openListener);
      }
    }

    this.socket = socket;
    this.socket.addEventListener('message', messageListener);
    this.socket.addEventListener('open', openListener);

    if (this.socket.binaryType === 'blob') {
      this.socket.binaryType = 'arraybuffer';
    }

    return this;
  };

  stream._write = function(chunk, encoding, callback) {
    if (!isSocketOpen()) {
      pendingWrite = { chunk: chunk, callback: callback };
      return;
    }

    socketSend(chunk, callback);
  };

  stream._read = function(size) {};

  return stream;
};
{'fraction_non_alphanumeric': 0.13633879781420766, 'fraction_numerical': 0.001639344262295082, 'mean_word_length': 2.8415529905561385, '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': '7329444', 'n_tokens_mistral': 1086, 'n_tokens_neox': 1043, 'n_words': 221}