<commit_before>import tls from 'tls';
import crypto from 'crypto';
import { EventEmitter } from 'events';

import {} from './buffertools';
import { Packet, TYPE, HEADER_LENGTH as packetHeaderLength } from './packet';
import StreamParser from './stream-parser';

class ReadablePacketStream extends StreamParser {
  *parser() {
    for (;;) {
      const header = yield this.readBuffer(packetHeaderLength);
      const length = header.readUInt16BE(2);
      const data = yield this.readBuffer(length - packetHeaderLength);

      this.push(new Packet(Buffer.concat([header, data])));
    }
  }
}

export default class MessageIO extends EventEmitter {
  constructor(socket, _packetSize, debug) {
    super();

    this.socket = socket;
    this._packetSize = _packetSize;
    this.debug = debug;
    this.sendPacket = this.sendPacket.bind(this);

    this.packetStream = new ReadablePacketStream();
    this.packetStream.on('data', (packet) => {
      this.logPacket('Received', packet);
      this.emit('data', packet.data());
      if (packet.isLast()) {
        this.emit('message');
      }
    });

    this.socket.pipe(this.packetStream);
    this.packetDataSize = this._packetSize - packetHeaderLength;<commit_msg>Rewrite the Readable Packet Stream to not use generators.<commit_after>import tls from 'tls';
import crypto from 'crypto';
import { EventEmitter } from 'events';
import { Transform } from 'readable-stream';

import {} from './buffertools';
import { Packet, TYPE, HEADER_LENGTH as packetHeaderLength } from './packet';

class ReadablePacketStream extends Transform {
  constructor() {
    super({ "objectMode": true });

    this.buffer = new Buffer(0);
  }

  _transform(chunk, encoding, callback) {
    if (this.buffer.length) {
      this.buffer = Buffer.concat([this.buffer, chunk], this.buffer.length + chunk.length);
    } else {
      this.buffer = chunk;
    }

    // We need to have at least 4 bytes available to be able to determine
    // the full packet size.
    while (this.buffer.length >= 4) {
      const length = this.buffer.readUInt16BE(2);

      if (this.buffer.length >= length) {
        const data = this.buffer.slice(0, length);
        this.buffer = this.buffer.slice(length);
        this.push(new Packet(data));
      } else {
        // Not enough data to provide the next packet. Stop here and wait for
        // the next call to `_transform`.
        break;
      }
    }

    callback();
  }
}

export default class MessageIO extends EventEmitter {
  constructor(socket, _packetSize, debug) {
    super();

    this.socket = socket;
    this._packetSize = _packetSize;
    this.debug = debug;
    this.sendPacket = this.sendPacket.bind(this);

    this.packetStream = new ReadablePacketStream();
    this.packetStream.on('data', (packet) => {
      this.logPacket('Received', packet);
      this.emit('data', packet.data());
      if (packet.isLast()) {
        this.emit('message');
      }
    });

    this.socket.pipe(this.packetStream);
    this.packetDataSize = this._packetSize - packetHeaderLength;
{'fraction_non_alphanumeric': 0.1295387634936212, 'fraction_numerical': 0.0032711808963035655, 'mean_word_length': 3.1492537313432836, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 7, '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': '4366195', 'n_tokens_mistral': 893, 'n_tokens_neox': 870, 'n_words': 230}