<commit_before>var assert  = require('assert');
var utils   = require('./utils');
var logging = require('./logging');

var MAX_HTTP_PAYLOAD_SIZE = 16383; // TODO: this is repeated in multiple files

// The Stream class
// ================

// Stream is a [Duplex stream](http://nodejs.org/api/stream.html#stream_class_stream_duplex)
// subclass that implements the [HTTP/2 Stream](http://http2.github.io/http2-spec/#rfc.section.3.4)
// concept. It has two 'sides': one that is used by the user to send/receive data (the `stream`
// object itself) and one that is used by a Connection to read/write frames to/from the other peer
// (`stream.upstream`).

var Duplex = require('stream').Duplex;

exports.Stream = Stream;

// Public API
// ----------

// Public events, methods and fields, with arguments listed in the numbered list:
//
// * **Event: 'headers'**: signals an incoming HEADERS frame
//   1. a map of the header names and values
// * **Event: 'promise'**: signals an incoming PUSH_PROMISE frame
//   1. a map of the promise header names and values
//   2. a reference to the promised stream
// * **Event: 'error'**: signals an error
//   1. upper cased type of the error (PROTOCOL_ERROR, etc.)
// * **headers(headers, [priority])**: send a HEADERS frame
//   1. a map of the header names and values
//   2. priority value to be sent (optional)
// * **promise(headers, stream)**: promise a stream
//   1. a map of the promise header names and values
//   2. a reference to the promised stream
// * **reset(error)**: reset the stream with an error code
//   1. the name of the error
// * **setInitialWindow(size)**: when creating the stream, the initial flow control window size
//   is 65535. It can be changed *any time* ([as described in the standard][2]) using this method.
//   1. the new value of the initial window size
// * **upstream**: a duplex stream that is used by the parent connection to write/read frames
//   that are to be sent/arrived to/from the peer and are related to this stream.
//
// [1]: http://nodejs.org/api/http.html#http_message_headers
// [2]: http://tools.ietf.org/html/draft-ietf-httpbis-http2-04#section-6.9.2

// Constructor
// -----------

// The main aspects of managing the stream are:
function Stream(log) {
  Duplex.call(this);

  // * every method uses the common logger object
  this._log = (log || logging.root).child({ component: 'stream' });

  // * receiving and sending stream management commands
  this._initializeManagement();

  // * sending and receiving frames to/from the upstream connection
  this._initializeDataFlow();

  // * maintaining the state of the stream (idle, open, closed, etc.) and error detection
  this._initializeState();
}

Stream.prototype = Object.create(Duplex.prototype, { constructor: { value: Stream } });

// Managing the stream
// -------------------

// PUSH_PROMISE and HEADERS are forwarded to the user through events. When error happens, we first
// close the stream.
Stream.prototype._initializeManagement = function _initializeManagement() {
  this.on('receiving', function(frame) {
    if (frame.type === 'PUSH_PROMISE') {
      this.emit('promise', frame.headers, frame.promised_stream);
    } else if (frame.type === 'HEADERS') {
      this.priority = frame.priority;
      this.emit('headers', frame.headers);
    }
  });
};

// For sending management frames, the `this.upstream.push(frame)` method is used. It notifies the state
// management code about the sent frames (using the 'sending' event) so we don't have to manage
// state transitions here.
Stream.prototype.promise = function promise(stream, headers) {
  stream.emit('promise_initiated');
  this.upstream.push({
    type: 'PUSH_PROMISE',
    promised_stream: stream,
    headers: headers
  });
};

Stream.prototype.headers = function headers(headers, priority) {
  this.upstream.push({
    type: 'HEADERS',
    priority: priority,
    headers: headers
  });
  this.priority = priority;
};

Stream.prototype.reset = function reset(error) {
  this.upstream.push({
    type: 'RST_STREAM',
    error: error
  });
};

// Data flow
// ---------

var Flow = require('./flow').Flow;

// The incoming and the generated outgoing frames are received/transmitted on the `this.upsteam`
// [Flow](flow.html). The [Connection](connection.html) object instantiating the stream will read
// and write frames to/from it.
Stream.prototype._initializeDataFlow = function _initializeDataFlow() {
  this.on('finish', this._finishing.bind(this));

  this.upstream = new Flow(this._log);
  this.upstream._send = this._send.bind(this);
  this.upstream._receive = this._receive.bind(this);
  this.upstream.on('sending', this.emit.bind(this, 'sending'));
  this.upstream.on('receiving', this.emit.bind(this, 'receiving'));
};

// Handling an incoming frame:
Stream.prototype._receive = function _receive(frame, ready) {
  var callReady = true;

  // * If its a DATA frame
  if (frame.type === 'DATA') {
    var moreNeeded = this.push(frame.data);
    if (!moreNeeded) {
      this._receiveMore = ready;
      callReady = false;
    }
  }

  // * Otherwise
  else {
    this.emit(frame.type, frame);
  }

  // * Any frame may signal the end of the stream with the END_STREAM flag
  if (frame.flags.END_STREAM) {
    this.push(null);
  }

  if (callReady) {
    ready();
  }
};

Stream.prototype._read = function _read() {
  if (this._receiveMore) {
    var receiveMore = this._receiveMore;
    delete this._receiveMore;
    receiveMore();
  }
};

// Handling a write request from the user:
Stream.prototype._write = function _write(buffer, encoding, ready) {
  // * The incoming buffer is cut into pieces that are not larger than `MAX_HTTP_PAYLOAD_SIZE`
  var chunks = utils.cut(buffer, MAX_HTTP_PAYLOAD_SIZE);

  // * Chunks are wrapped in DATA frames and sent out until all of them are sent or the flow control
  //   `window` is not enough to send a chunk
  var moreNeeded = true;
  while (chunks.length > 0) {
    moreNeeded = this.upstream.push({
      type: 'DATA',
      flags: {},
      data: chunks.shift()
    });
  }

  if (moreNeeded) {
    ready();
  } else {
    this._sendMore = ready;
  }
};<commit_msg>Stream: smarter chunking heuristic. Fixes #25.<commit_after>var assert  = require('assert');
var utils   = require('./utils');
var logging = require('./logging');

var MAX_HTTP_PAYLOAD_SIZE = 16383; // TODO: this is repeated in multiple files

// The Stream class
// ================

// Stream is a [Duplex stream](http://nodejs.org/api/stream.html#stream_class_stream_duplex)
// subclass that implements the [HTTP/2 Stream](http://http2.github.io/http2-spec/#rfc.section.3.4)
// concept. It has two 'sides': one that is used by the user to send/receive data (the `stream`
// object itself) and one that is used by a Connection to read/write frames to/from the other peer
// (`stream.upstream`).

var Duplex = require('stream').Duplex;

exports.Stream = Stream;

// Public API
// ----------

// Public events, methods and fields, with arguments listed in the numbered list:
//
// * **Event: 'headers'**: signals an incoming HEADERS frame
//   1. a map of the header names and values
// * **Event: 'promise'**: signals an incoming PUSH_PROMISE frame
//   1. a map of the promise header names and values
//   2. a reference to the promised stream
// * **Event: 'error'**: signals an error
//   1. upper cased type of the error (PROTOCOL_ERROR, etc.)
// * **headers(headers, [priority])**: send a HEADERS frame
//   1. a map of the header names and values
//   2. priority value to be sent (optional)
// * **promise(headers, stream)**: promise a stream
//   1. a map of the promise header names and values
//   2. a reference to the promised stream
// * **reset(error)**: reset the stream with an error code
//   1. the name of the error
// * **setInitialWindow(size)**: when creating the stream, the initial flow control window size
//   is 65535. It can be changed *any time* ([as described in the standard][2]) using this method.
//   1. the new value of the initial window size
// * **upstream**: a duplex stream that is used by the parent connection to write/read frames
//   that are to be sent/arrived to/from the peer and are related to this stream.
//
// [1]: http://nodejs.org/api/http.html#http_message_headers
// [2]: http://tools.ietf.org/html/draft-ietf-httpbis-http2-04#section-6.9.2

// Constructor
// -----------

// The main aspects of managing the stream are:
function Stream(log) {
  Duplex.call(this);

  // * every method uses the common logger object
  this._log = (log || logging.root).child({ component: 'stream' });

  // * receiving and sending stream management commands
  this._initializeManagement();

  // * sending and receiving frames to/from the upstream connection
  this._initializeDataFlow();

  // * maintaining the state of the stream (idle, open, closed, etc.) and error detection
  this._initializeState();
}

Stream.prototype = Object.create(Duplex.prototype, { constructor: { value: Stream } });

// Managing the stream
// -------------------

// PUSH_PROMISE and HEADERS are forwarded to the user through events. When error happens, we first
// close the stream.
Stream.prototype._initializeManagement = function _initializeManagement() {
  this.on('receiving', function(frame) {
    if (frame.type === 'PUSH_PROMISE') {
      this.emit('promise', frame.headers, frame.promised_stream);
    } else if (frame.type === 'HEADERS') {
      this.priority = frame.priority;
      this.emit('headers', frame.headers);
    }
  });
};

// For sending management frames, the `this.upstream.push(frame)` method is used. It notifies the state
// management code about the sent frames (using the 'sending' event) so we don't have to manage
// state transitions here.
Stream.prototype.promise = function promise(stream, headers) {
  stream.emit('promise_initiated');
  this.upstream.push({
    type: 'PUSH_PROMISE',
    promised_stream: stream,
    headers: headers
  });
};

Stream.prototype.headers = function headers(headers, priority) {
  this.upstream.push({
    type: 'HEADERS',
    priority: priority,
    headers: headers
  });
  this.priority = priority;
};

Stream.prototype.reset = function reset(error) {
  this.upstream.push({
    type: 'RST_STREAM',
    error: error
  });
};

// Data flow
// ---------

var Flow = require('./flow').Flow;

// The incoming and the generated outgoing frames are received/transmitted on the `this.upsteam`
// [Flow](flow.html). The [Connection](connection.html) object instantiating the stream will read
// and write frames to/from it.
Stream.prototype._initializeDataFlow = function _initializeDataFlow() {
  this.on('finish', this._finishing.bind(this));

  this.upstream = new Flow(this._log);
  this.upstream._send = this._send.bind(this);
  this.upstream._receive = this._receive.bind(this);
  this.upstream.on('sending', this.emit.bind(this, 'sending'));
  this.upstream.on('receiving', this.emit.bind(this, 'receiving'));
};

// Handling an incoming frame:
Stream.prototype._receive = function _receive(frame, ready) {
  var callReady = true;

  // * If its a DATA frame
  if (frame.type === 'DATA') {
    var moreNeeded = this.push(frame.data);
    if (!moreNeeded) {
      this._receiveMore = ready;
      callReady = false;
    }
  }

  // * Otherwise
  else {
    this.emit(frame.type, frame);
  }

  // * Any frame may signal the end of the stream with the END_STREAM flag
  if (frame.flags.END_STREAM) {
    this.push(null);
  }

  if (callReady) {
    ready();
  }
};

Stream.prototype._read = function _read() {
  if (this._receiveMore) {
    var receiveMore = this._receiveMore;
    delete this._receiveMore;
    receiveMore();
  }
};

// Handling a write request from the user:
Stream.prototype._write = function _write(buffer, encoding, ready) {
  // * The incoming buffer is cut into pieces that are no larger than `MAX_HTTP_PAYLOAD_SIZE` and
  //   the initial window size, but at least 100 bytes. Ideal chunk size is such that most chunks
  //   are of equal size. This is a mostly heuristic algorithm, and should be changed to a more
  //   fine-grained chunking method in the future.
  var limit = Math.max(100, Math.min(MAX_HTTP_PAYLOAD_SIZE, this.upstream._initialWindow));
  var idealChunkCount = Math.ceil(buffer.length / limit);
  var chunkSize = Math.round(buffer.length / idealChunkCount);
  var chunks = utils.cut(buffer, chunkSize);

  // * Chunks are wrapped in DATA frames and sent out until all of them are sent or the flow control
  //   `window` is not enough to send a chunk
  var moreNeeded = true;
  while (chunks.length > 0) {
    moreNeeded = this.upstream.push({
      type: 'DATA',
      flags: {},
      data: chunks.shift()
    });
  }

  if (moreNeeded) {
    ready();
  } else {
    this._sendMore = ready;
  }
};
{'fraction_non_alphanumeric': 0.13890192397935242, 'fraction_numerical': 0.006100422336931018, 'mean_word_length': 3.308288409703504, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '4065783', 'n_tokens_mistral': 3900, 'n_tokens_neox': 3790, 'n_words': 1293}