http2 - npm Package Compare versions

Comparing version 0.1.0 to 0.1.1

example/client.js

@@ -5,7 +5,2 @@ var parse_url = require('url').parse;
 
-var settings = {
-  SETTINGS_MAX_CONCURRENT_STREAMS: 1,
-  SETTINGS_INITIAL_WINDOW_SIZE: 100000
-};
-
 var url = parse_url(process.argv.pop());
@@ -12,0 +7,0 @@

HISTORY.md

 Version history
 ===============
 
+### 0.1.1 (2013-08-12) ###
+
+* Lots of bugfixes
+* Proper flow control for outgoing frames
+* Basic flow control for incoming frames
+* [Blog post](http://gabor.molnar.es/blog/2013/08/12/gsoc-week-number-8/)
+* [Tarball](https://github.com/molnarg/node-http2/archive/node-http2-0.1.1.tar.gz)
+
 ### 0.1.0 (2013-08-06) ###
@@ -5,0 +13,0 @@

lib/compressor.js

@@ -16,3 +16,2 @@ // HTTP/2 compression is implemented by two [Transform Stream][1] subclasses that operate in
 
-var utils = require('../lib/utils');
 var logging = require('./logging');
@@ -371,3 +370,3 @@ var Transform = require('stream').Transform;
 
-  var buffer = utils.concat(Array.prototype.concat.apply([], buffers)); // [[bufs]] -> [bufs] -> buf
+  var buffer = concat(Array.prototype.concat.apply([], buffers)); // [[bufs]] -> [bufs] -> buf
 
@@ -637,3 +636,3 @@ this._log.trace({ data: buffer }, 'Header compression is done');
     // * cuts the header block into `chunks` that are not larger than `MAX_HTTP_PAYLOAD_SIZE`
-    var chunks = utils.cut(buffer, MAX_HTTP_PAYLOAD_SIZE);
+    var chunks = cut(buffer, MAX_HTTP_PAYLOAD_SIZE);
 
@@ -644,3 +643,3 @@ // * for each chunk, it pushes out a chunk frame that is identical to the original, except
     for (var i = 0; i < chunks.length; i++) {
-      var flags = utils.shallowCopy(frame.flags);
+      var flags = shallowCopy(frame.flags);
       if (i === chunks.length - 1) {
@@ -714,3 +713,3 @@ flags['END_' + frame.type] = true;
   if (this._inProgress && (frame.flags.END_HEADERS || frame.flags.END_PUSH_PROMISE)) {
-    var buffer = utils.concat(this._frames.map(function(frame) {
+    var buffer = concat(this._frames.map(function(frame) {
       return frame.data;
@@ -813,1 +812,40 @@ }));
 ];
+
+// Helper functions
+// ----------------
+
+// Concatenate an array of buffers into a new buffer
+function concat(buffers) {
+  var size = 0;
+  for (var i = 0; i < buffers.length; i++) {
+    size += buffers[i].length;
+  }
+
+  var concatenated = new Buffer(size);
+  for (var cursor = 0, j = 0; j < buffers.length; cursor += buffers[j].length, j++) {
+    buffers[j].copy(concatenated, cursor);
+  }
+
+  return concatenated;
+}
+
+// Cut `buffer` into chunks not larger than `size`
+function cut(buffer, size) {
+  var chunks = [];
+  var cursor = 0;
+  do {
+    var chunkSize = Math.min(size, buffer.length - cursor);
+    chunks.push(buffer.slice(cursor, cursor + chunkSize));
+    cursor += chunkSize;
+  } while(cursor < buffer.length);
+  return chunks;
+}
+
+// Shallow copy inspired by underscore's [clone](http://underscorejs.org/#clone)
+function shallowCopy(object) {
+  var clone = {};
+  for (var key in object) {
+    clone[key] = object[key];
+  }
+  return clone;
+}

lib/connection.js

@@ -1,23 +0,39 @@
-var assert  = require('assert');
-var utils   = require('./utils');
 var logging = require('./logging');
 
-// Connection
-// ----------
+// The Connection class
+// ====================
 
 // The Connection class manages HTTP/2 connections. Each instance corresponds to one transport
-// stream (TCP stream). It operates by sending and receiving frames and is implemented as an
-// [object mode][1] [Duplex stream][2].
-//
-// [1]: http://nodejs.org/api/stream.html#stream_new_stream_readable_options
-// [2]: http://nodejs.org/api/stream.html#stream_class_stream_duplex
+// stream (TCP stream). It operates by sending and receiving frames and is implemented as a
+// [Flow](flow.html) subclass.
 
-var Duplex  = require('stream').Duplex;
+var Flow = require('./flow').Flow;
 
 exports.Connection = Connection;
 
+// Public API
+// ----------
+
+// * **new Flow(firstStreamId, settings, [log])**: create a new Connection
+//
+// * **Event: 'error' (type)**: signals a connection level error
+//
+// * **Event: 'stream' (stream)**: signals that there's an incoming stream
+//
+// * **createStream(): stream**: initiate a new stream
+//
+// * **set(settings)**: change the value of one or more settings according to the key-value pairs
+//   of `settings`
+//
+// * **ping(callback)**: send a ping and call callback when the answer arrives
+//
+// * **close([error])**: close the stream with an error code
+
+// Constructor
+// -----------
+
 // The main aspects of managing the connection are:
 function Connection(firstStreamId, settings, log) {
-  // * handling IO, particularly multiplexing/demultiplexing incoming and outgoing frames
-  Duplex.call(this, { objectMode: true });
+  // * initializing the base class
+  Flow.call(this, 0);
 
@@ -38,4 +54,6 @@ // * logging: every method uses the common logger object
   this._initializeFlowControl();
+
+  // * multiplexing
 }
-Connection.prototype = Object.create(Duplex.prototype, { constructor: { value: Connection } });
+Connection.prototype = Object.create(Flow.prototype, { constructor: { value: Connection } });
 
@@ -90,11 +108,8 @@ // Overview
 
-  // * The next outbound stream ID is stored in `this._nextStreamId`
+  // * The next outbound stream ID and the last inbound stream id
   this._nextStreamId = firstStreamId;
+  this._lastIncomingStream = 0;
 
-  // * Creating the `_control` stream that corresponds to stream ID 0 (connection level frames).
-  this._control = new Duplex({ objectMode: true });
-  this._control._write = this._writeControlFrame.bind(this);
-  this._control._read = utils.noop;
-  this._control.on('readable', this.emit.bind(this, 'stream_readable'));
-  this._streamsIds[0] = this._streamPriorities[0] = { upstream: this._control, priority: -1 };
+  // * Calling `_writeControlFrame` when there's an incoming stream with 0 as stream ID
+  this._streamsIds[0] = { upstream: { write: this._writeControlFrame.bind(this) } };
 
@@ -105,6 +120,6 @@ // * By default, the number of concurrent outbound streams is not limited. The `_streamLimit` can
   this._streamLimit = Infinity;
-  this._control.on('SETTINGS_MAX_CONCURRENT_STREAMS', this._updateStreamLimit.bind(this));
+  this.on('SETTINGS_MAX_CONCURRENT_STREAMS', this._updateStreamLimit);
 };
 
-Connection.prototype.getIdOf = function getIdOf(stream) {
+Connection.prototype._getIdOf = function _getIdOf(stream) {
   return this._streamsIds.indexOf(stream);
@@ -115,5 +130,9 @@ };
 // broadcasts the message by creating an event on it.
-Connection.prototype._writeControlFrame = function _writeControlFrame(frame, encoding, done) {
-  this._control.emit(frame.type, frame);
-  done();
+Connection.prototype._writeControlFrame = function _writeControlFrame(frame) {
+  if ((frame.type === 'SETTINGS') || (frame.type === 'PING') ||
+      (frame.type === 'GOAWAY') || (frame.type === 'WINDOW_UPDATE')) {
+    this.emit(frame.type, frame);
+  } else {
+    this.emit('error', 'PROTOCOL_ERROR');
+  }
 };
@@ -143,4 +162,5 @@
   this._log.trace({ id: id }, 'Adding new stream.');
-  var stream = new Stream(this._log.child({ stream: id }));
+  var stream = new Stream(this._log.child({ stream_id: id }));
   this._streamsIds[id] = stream;
+  this.emit('new_stream', stream, id);
   return stream;
@@ -150,5 +170,5 @@ };
 Connection.prototype._activateStream = function _activateStream(stream) {
-  this._log.trace({ id: this.getIdOf(stream) }, 'Activating stream.');
+  this._log.trace({ id: this._getIdOf(stream) }, 'Activating stream.');
   this._streamPriorities.push(stream);
-  stream.upstream.on('readable', this.emit.bind(this, 'stream_readable'));
+  stream.upstream.on('readable', this.read.bind(this, 0));
 };
@@ -159,6 +179,16 @@
 //
-// * creating and activating the stream
-// * emitting 'stream' event with the new stream
-Connection.prototype._incomingStream = function _incomingStream(id) {
+// * Incoming stream IDs have to be greater than any previous incoming stream ID, and have to be of
+//   different parity than IDs used for outbound streams.
+// * It creates and activates the stream.
+// * Emits 'stream' event with the new stream.
+Connection.prototype._createIncomingStream = function _createIncomingStream(id) {
   this._log.debug({ id: id }, 'New incoming stream.');
+
+  if ((id <= this._lastIncomingStream) || ((id - this._nextStreamId) % 2 === 0)) {
+    this._log.error({ id: id, lastIncomingStream: this._lastIncomingStream }, 'Invalid incoming stream ID.');
+    this.emit('error', 'PROTOCOL_ERROR');
+    return undefined;
+  }
+
+  this._lastIncomingStream = id;
   var stream = this._newStream(id);
@@ -220,13 +250,14 @@ this._activateStream(stream);
 
-// The `_read` method is a [virtual method of the Duplex class][1] that has to be implemented by
-// child classes. It reads frames from streams and pushes them to the output buffer.
-// [1]: http://nodejs.org/api/stream.html#stream_readable_read_size
-Connection.prototype._read = function _read() {
+// The `_send` method is a virtual method of the [Flow class](flow.html) that has to be implemented
+// by child classes. It reads frames from streams and pushes them to the output buffer.
+Connection.prototype._send = function _send() {
   this._log.trace('Starting forwarding frames from streams.');
 
-  // * Looping through the active streams in priority order, forwarding until:
-  var moreNeeded = true, stream, id, frame;
-  for (var i = 0; i < this._streamPriorities.length && moreNeeded; i++) {
-    stream = this._streamPriorities[i];
-    id = this.getIdOf(stream);
+  // * Looping through the active streams in priority order and forwarding frames from streams
+  stream_loop:
+  for (var i = 0; i < this._streamPriorities.length; i++) {
+    var stream = this._streamPriorities[i];
+    var id = this._getIdOf(stream);
+    var frame;
+    var unshiftRemainder = stream.upstream.unshift.bind(stream.upstream);
     while (frame = stream.upstream.read()) {
@@ -236,33 +267,19 @@ frame.stream = id;
         frame.promised_stream.emit('promise_sent');
-        frame.promised_stream = this.getIdOf(frame.promised_stream);
+        frame.promised_stream = this._getIdOf(frame.promised_stream);
       }
-      moreNeeded = this._send(frame);
-    }
-  }
 
-  // * there are no more frames in the buffers of the streams, but more would be needed
-  //   * coming back once a stream becomes readable again
-  if (i === this._streamPriorities.length) {
-    this._log.trace('More chunk is needed, but we could not provide more.');
-    this.once('stream_readable', this._read.bind(this));
-  }
+      var moreNeeded = this._push(frame, unshiftRemainder);
 
-  // * it's not possible to send more because of flow control
-  //   * coming back once flow control window is updated
-  else if (moreNeeded === null) {
-    this._log.trace('We could not send more because of insufficient flow control window.');
-    this.once('window_update', this._read.bind(this));
+      if (moreNeeded === null) {
+        continue stream_loop;
+      } else if (moreNeeded === false) {
+        break stream_loop;
+      }
+    }
   }
-
-  // * no more chunk needed
-  //   * coming back only when `_read` is called again by Duplex
-  else if (moreNeeded === false) {
-    this._log.trace('No more chunk needed, stopping forwarding.');
-  }
 };
 
-// The `_write` method is another [virtual method of the Duplex class][1] that has to be implemented
-// by child classes. It forwards the given frame to the appropriate stream:
-// [1]: http://nodejs.org/api/stream.html#stream_writable_write_chunk_encoding_callback
-Connection.prototype._write = function write(frame, encoding, done) {
+// The `_receive` method is another virtual method of the [Flow class](flow.html) that has to be
+// implemented by child classes. It forwards the given frame to the appropriate stream:
+Connection.prototype._receive = function _receive(frame, done) {
   // * gets the appropriate stream from the stream registry
@@ -273,3 +290,3 @@ var stream = this._streamsIds[frame.stream];
   if (!stream) {
-    stream = this._incomingStream(frame.stream);
+    stream = this._createIncomingStream(frame.stream);
   }
@@ -279,8 +296,5 @@
   if (frame.type === 'PUSH_PROMISE') {
-    frame.promised_stream = this._incomingStream(frame.promised_stream);
+    frame.promised_stream = this._createIncomingStream(frame.promised_stream);
   }
 
-  // * tells the world that there's an incoming frame
-  this.emit('receiving', frame);
-
   // * and writes it to the `stream`'s `upstream`
@@ -299,4 +313,2 @@ stream.upstream.write(frame);
   this._log.info('Sending the first SETTINGS frame as part of the connection header.');
-  assert('SETTINGS_MAX_CONCURRENT_STREAMS' in settings);
-  assert('SETTINGS_INITIAL_WINDOW_SIZE' in settings);
   this.set(settings);
@@ -315,3 +327,3 @@
   // * Forwarding SETTINGS frames to the `_receiveSettings` method
-  this._control.on('SETTINGS', this._receiveSettings.bind(this));
+  this.on('SETTINGS', this._receiveSettings);
 };
@@ -322,3 +334,3 @@
   for (var name in frame.settings) {
-    this._control.emit(name, frame.settings[name]);
+    this.emit(name, frame.settings[name]);
   }
@@ -329,3 +341,4 @@ };
 Connection.prototype.set = function set(settings) {
-  this._control.push({
+  this.push({
+    stream: 0,
     type: 'SETTINGS',
@@ -348,8 +361,4 @@ settings: settings
   this._pings = {};
-  this._control.on('PING', this._receivePing.bind(this));
-  this._control.on('GOAWAY', this._receiveGoaway.bind(this));
-  this._lastIncomingStream = 0;
-  this.on('stream', function(stream, id) {
-    this._lastIncomingStream = id;
-  }.bind(this));
+  this.on('PING', this._receivePing);
+  this.on('GOAWAY', this._receiveGoaway);
 };
@@ -375,3 +384,4 @@
   this._log.debug({ data: data }, 'Sending PING.');
-  this._control.push({
+  this.push({
+    stream: 0,
     type: 'PING',
@@ -381,3 +391,3 @@ flags: {
     },
-    data: new Buffer(id, 'hex')
+    data: data
   });
@@ -400,3 +410,4 @@ };
     this._log.debug({ data: frame.data }, 'Answering PING.');
-    this._control.push({
+    this.push({
+      stream: 0,
       type: 'PING',
@@ -413,3 +424,5 @@ flags: {
 Connection.prototype.close = function close(error) {
+  this._log.info({ error: error }, 'Closing the connection');
   this.push({
+    stream: 0,
     type: 'GOAWAY',
@@ -423,2 +436,3 @@ last_stream: this._lastIncomingStream,
 Connection.prototype._receiveGoaway = function _receiveGoaway(frame) {
+  this._log.info({ error: frame.error }, 'Other end closed the connection');
   this.push(null);
@@ -431,80 +445,45 @@ };
 Connection.prototype._initializeFlowControl = function _initializeFlowControl() {
-  // Turning off flow control for incoming frames (not yet supported).
-  this._control.push({
-    type: 'WINDOW_UPDATE',
-    flags: {
-      END_FLOW_CONTROL: true
-    },
-    window_size: 0
+  // Handling of initial window size of individual streams.
+  this._initialStreamWindowSize = INITIAL_STREAM_WINDOW_SIZE;
+  this.on('new_stream', function(stream) {
+    stream.upstream.setInitialWindow(this._initialStreamWindowSize);
   });
+  this.on('SETTINGS_INITIAL_WINDOW_SIZE', this._setInitialStreamWindowSize);
+  this.on('SETTINGS_FLOW_CONTROL_OPTIONS', this._setStreamFlowControl);
+  this._streamsIds[0].upstream.setInitialWindow = function noop() {};
 
-  // Initializing flow control for outgoing frames
-  this._window = INITIAL_WINDOW_SIZE;
-  this._control.on('WINDOW_UPDATE', this._updateWindow.bind(this));
+  // Flow control for incoming frames is not yet supported, and is turned off in the initial
+  // SETTINGS frame.
 };
 
-// When a HTTP/2.0 connection is first established, new streams are created with an initial flow
-// control window size of 65535 bytes.
-var INITIAL_WINDOW_SIZE = 65535;
+// The initial connection flow control window is 65535 bytes.
+var INITIAL_STREAM_WINDOW_SIZE = 65535;
 
 // A SETTINGS frame can alter the initial flow control window size for all current streams. When the
-// value of SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST adjust the size of all stream by
-// calling the `setInitialWindowSize` method. The window size has to be modified by the difference
-// between the new value and the old value.
-Connection.prototype.setInitialWindowSize = function setInitialWindowSize(initialWindowSize) {
-  this._window = this._window - this._initialWindowSize + initialWindowSize;
-  this._initialWindowSize = initialWindowSize;
-};
-
-// Flow control can be disabled for all streams on the connection using the `disableFlowControl`
-// method. This may happen when there's a SETTINGS frame received with the
-// SETTINGS_FLOW_CONTROL_OPTIONS setting.
-Connection.prototype.disableFlowControl = function disableFlowControl() {
-  this._window = Infinity;
-};
-
-// The `_updateWindow` method gets called every time there's an incoming WINDOW_UPDATE frame. It
-// modifies the modifies the flow control window:
-//
-// * Flow control can be disabled for an individual stream by sending a WINDOW_UPDATE with the
-//   END_FLOW_CONTROL flag set. The payload of a WINDOW_UPDATE frame that has the END_FLOW_CONTROL
-//   flag set is ignored.
-// * A sender that receives a WINDOW_UPDATE frame updates the corresponding window by the amount
-//   specified in the frame.
-Connection.prototype._updateWindow = function _updateWindow(frame) {
-  if (frame.flags.END_FLOW_CONTROL) {
-    this.disableFlowControl();
+// value of SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST adjust the window size of all
+// stream by calling the `setInitialStreamWindowSize` method. The window size has to be modified by
+// the difference between the new value and the old value.
+Connection.prototype._setInitialStreamWindowSize = function _setInitialStreamWindowSize(size) {
+  if ((this._initialStreamWindowSize === Infinity) && (size !== Infinity)) {
+    this._log.error('Trying to manipulate initial flow control window size after flow control was turned off.');
+    this.emit('error', 'FLOW_CONTROL_ERROR');
   } else {
-    this._window += frame.window_size;
+    this._log.debug({ size: size }, 'Changing stream initial window size.');
+    this._initialStreamWindowSize = size;
+    this._streamsIds.forEach(function(stream) {
+      stream.upstream.setInitialWindow(size);
+    });
   }
-  this.emit('window_update');
 };
 
-Connection.prototype._send = function _send(frame) {
-  if (frame && (frame.type === 'DATA')) {
-    if (frame.data.length > this._window) {
-      return null;
-    }
-    this._window -= frame.data.length;
+// `_setStreamFlowControl()` may be used to disable/enable flow control. In practice, it is just
+// for turning off flow control since it can not be turned on.
+Connection.prototype._setStreamFlowControl = function _setStreamFlowControl(disable) {
+  if (disable) {
+    this._increaseWindow(Infinity);
+    this._setInitialStreamWindowSize(Infinity);
+  } else if (this._initialStreamWindowSize === Infinity) {
+    this._log.error('Trying to re-enable flow control after it was turned off.');
+    this.emit('error', 'FLOW_CONTROL_ERROR');
   }
-
-  return this.push(frame);
 };
-
-// TODO list
-// ---------
-//
-// * Stream management
-//   * check if the stream initiated by the peer has a stream id with appropriate parity
-//   * check for invalid frame types on the control stream
-//   * _activateStream:
-//     * respect priority when inserting
-// * Multiplexing
-//   * prioritization
-//   * if we are on the flow control limit, it's still possible to send non-DATA frames
- // * Settings management
-//   * storing and broadcasting the incoming settings
-// * Lifecycle management
-//   * implementing connection tear down procedure
-// * Flow control
-//   * setting the initial window size of streams (based on SETTINGS_INITIAL_WINDOW_SIZE)

lib/endpoint.js

@@ -139,13 +139,11 @@ var logging      = require('./logging');
   this._deserializer.pipe(this._decompressor).pipe(this._connection);
+
+  this._serializer.on('readable', this._read.bind(this));
 };
 
-Endpoint.prototype._read = function _read(size) {
+Endpoint.prototype._read = function _read() {
   var moreNeeded = true, chunk;
-  while (moreNeeded && (chunk = this._serializer.read(size))) {
+  while (moreNeeded && (chunk = this._serializer.read())) {
     moreNeeded = this.push(chunk);
   }
-
-  if (moreNeeded) {
-    this._serializer.once('readable', this._read.bind(this));
-  }
 };
@@ -152,0 +150,0 @@

lib/framer.js

@@ -121,3 +121,3 @@ // The framer consists of two [Transform Stream][1] subclasses that operate in [object mode][2]:
           this._log.error({ err: error }, 'Incoming frame parsing error');
-          this.emit('error', error);
+          this.emit('error', 'PROTOCOL_ERROR');
         }
@@ -310,3 +310,4 @@ } else {
     var buffer = new Buffer(4);
-    buffer.writeUInt32BE(frame.priority & 0x7fffffff, 0);
+    assert((0 <= frame.priority) && (frame.priority <= 0xffffffff));
+    buffer.writeUInt32BE(frame.priority, 0);
     buffers.push(buffer);
@@ -381,3 +382,5 @@ }
   var buffer = new Buffer(4);
-  buffer.writeUInt32BE(errorCodes.indexOf(frame.error), 0);
+  var code = errorCodes.indexOf(frame.error);
+  assert((0 <= code) && (code <= 0xffffffff));
+  buffer.writeUInt32BE(code, 0);
   buffers.push(buffer);
@@ -508,3 +511,4 @@ };
   var buffer = new Buffer(4);
-  buffer.writeUInt32BE(frame.promised_stream & 0x7fffffff, 0);
+  assert((0 <= frame.promised_stream) && (frame.promised_stream <= 0x7fffffff));
+  buffer.writeUInt32BE(frame.promised_stream, 0);
   buffers.push(buffer);
@@ -580,4 +584,10 @@ buffers.push(frame.data);
   var buffer = new Buffer(8);
-  buffer.writeUInt32BE(frame.last_stream & 0x7fffffff, 0);
-  buffer.writeUInt32BE(errorCodes.indexOf(frame.error), 4);
+
+  assert((0 <= frame.last_stream) && (frame.last_stream <= 0x7fffffff));
+  buffer.writeUInt32BE(frame.last_stream, 0);
+
+  var code = errorCodes.indexOf(frame.error);
+  assert((0 <= code) && (code <= 0xffffffff));
+  buffer.writeUInt32BE(code, 4);
+
   buffers.push(buffer);
@@ -615,3 +625,4 @@ };
   var buffer = new Buffer(4);
-  buffer.writeUInt32BE(frame.window_size & 0x7fffffff, 0);
+  assert((0 <= frame.window_size) && (frame.window_size <= 0x7fffffff));
+  buffer.writeUInt32BE(frame.window_size, 0);
   buffers.push(buffer);
@@ -655,3 +666,7 @@ };
   if (frame.data instanceof Buffer) {
-    logEntry.data = frame.data.toString('hex');
+    if (logEntry.data.length > 50) {
+      logEntry.data = frame.data.slice(0, 47).toString('hex') + '...';
+    } else {
+      logEntry.data = frame.data.toString('hex');
+    }
   }
@@ -658,0 +673,0 @@

lib/http.js

@@ -22,7 +22,6 @@ var tls = require('tls');
 
-// This should hold sane defaults for mandatory settings. These can be overridden by the user
-// using the options configuration object in client and server APIs.
+// This should hold sane defaults. These can be overridden by the user using the options
+// configuration object in client and server APIs.
 var default_settings = {
-  SETTINGS_MAX_CONCURRENT_STREAMS: 100,
-  SETTINGS_INITIAL_WINDOW_SIZE: 100000
+  SETTINGS_MAX_CONCURRENT_STREAMS: 100
 };
@@ -29,0 +28,0 @@

lib/index.js

@@ -7,5 +7,2 @@ // [node-http2](https://github.com/molnarg/node-http2) consists of the following components:
 
-// * [utils.js](utils.html): common utility functions, like concatenating buffers
-http2.utils      = require('./utils');
-
 // * [logging.js](logging.html): a default logger object and a registry of log formatter functions
@@ -12,0 +9,0 @@ http2.logging    = require('./logging');

lib/logging.js

@@ -1,2 +0,1 @@
-var utils = require('./utils');
 var logging = exports;
@@ -16,9 +15,10 @@
 } else {
+  function noop() {}
   logging.root = {
-    fatal: utils.noop,
-    error: utils.noop,
-    warn : utils.noop,
-    info : utils.noop,
-    debug: utils.noop,
-    trace: utils.noop,
+    fatal: noop,
+    error: noop,
+    warn : noop,
+    info : noop,
+    debug: noop,
+    trace: noop,
 
@@ -25,0 +25,0 @@ child: function() { return this; }

lib/stream.js

 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.
+// 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`).
 
@@ -15,2 +17,26 @@ var Duplex = require('stream').Duplex;
 
+// Public API
+// ----------
+
+// * **Event: 'headers' (headers)**: signals incoming headers
+//
+// * **Event: 'promise' (headers, stream)**: signals an incoming push promise
+//
+// * **Event: 'error' (type)**: signals an error
+//
+// * **headers(headers, [priority])**: send headers
+//
+// * **promise(headers, stream)**: promise a stream
+//
+// * **reset(error)**: reset the stream with an error code
+//
+// * **upstream**: a [Flow](flow.js) 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.
+//
+// Headers are always in the [regular node.js header format][1].
+// [1]: http://nodejs.org/api/http.html#http_message_headers
+
+// Constructor
+// -----------
+
 // The main aspects of managing the stream are:
@@ -23,14 +49,10 @@ function Stream(log) {
 
-  // * sending and receiving frames to/from the upstream connection
-  this._initializeUpstream();
-
   // * 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();
-
-  // * flow control, which includes forwarding data from/to the user on the Duplex stream interface
-  // (`write()`, `end()`, `pipe()`)
-  this._initializeFlowControl();
 }
@@ -46,16 +68,12 @@
 Stream.prototype._initializeManagement = function _initializeManagement() {
-  this.upstream.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);
-    }
-  }.bind(this));
-  this.on('error', function() {
-    this.push(null);
+  this.on('PUSH_PROMISE', function(frame) {
+    this.emit('promise', frame.headers, frame.promised_stream);
   });
+  this.on('HEADERS', function(frame) {
+    this.priority = frame.priority;
+    this.emit('headers', frame.headers);
+  });
 };
 
-// For sending management frames, the `this._send(frame)` method is used. It notifies the state
+// 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
@@ -65,3 +83,3 @@ // state transitions here.
   stream.emit('promise_initiated');
-  this._send({
+  this.upstream.push({
     type: 'PUSH_PROMISE',
@@ -74,3 +92,3 @@ promised_stream: stream,
 Stream.prototype.headers = function headers(headers, priority) {
-  this._send({
+  this.upstream.push({
     type: 'HEADERS',
@@ -84,3 +102,3 @@ priority: priority,
 Stream.prototype.reset = function reset(error) {
-  this._send({
+  this.upstream.push({
     type: 'RST_STREAM',
@@ -91,63 +109,133 @@ error: error
 
-// Managing the upstream connection
-// --------------------------------
+// Data flow
+// ---------
 
 // The incoming and the generated outgoing frames are received/transmitted on the `this.upsteam`
-// Duplex stream which operates in [object mode][1]. The [Connection](connection.html) object
-// instantiating the stream will read and write frames to/from it.
-// [1]: http://nodejs.org/api/stream.html#stream_new_stream_readable_options
-Stream.prototype._initializeUpstream = function _initializeUpstream() {
-  this._flushTimer = undefined;
-  this.on('finish', this._finishing.bind(this));
+// [Flow](flow.html). The [Connection](connection.html) object instantiating the stream will read
+// and write frames to/from it. The stream itself is a regular [Duplex stream][1], and is used by
+// the user to write or read the body of the request.
+// [1]: http://nodejs.org/api/stream.html#stream_class_stream_duplex
 
-  this.upstream = new Duplex({ objectMode: true });
-  this.upstream._queue = [];
-  this.upstream._read = utils.noop;
+//     upstream side                  stream                  user side
+//
+//                    +------------------------------------+
+//                    |                                    |
+//                    +------------------+                 |
+//                    |     upstream     |                 |
+//                    |                  |                 |
+//                    +--+               |              +--|
+//            read()  |  |  _send()      |    _write()  |  |  write(buf)
+//     <--------------|B |<--------------|--------------| B|<------------
+//                    |  |               |              |  |
+//            frames  +--+               |              +--|  buffers
+//                    |  |               |              |  |
+//     -------------->|B |---------------|------------->| B|------------>
+//      write(frame)  |  |  _receive()   |     _read()  |  |  read()
+//                    +--+               |              +--|
+//                    |                  |                 |
+//                    |                  |                 |
+//                    +------------------+                 |
+//                    |                                    |
+//                    +------------------------------------+
+//
+//     B: input or output buffer
 
-  // When there's an incoming frame, we let the world know this by emitting a 'receiving' event.
-  var log = this._log;
-  this.upstream._write = function(frame, encoding, done) {
-    log.debug({ frame: frame }, 'Receiving frame');
-    this.emit('receiving', frame);
-    done();
-  };
+var Flow = require('./flow').Flow;
+
+Stream.prototype._initializeDataFlow = function _initializeDataFlow() {
+  this.upstream = new Flow();
+  this.upstream._log = 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'));
+  this.upstream.on('error', this.emit.bind(this, 'error'));
+
+  this.on('finish', this._finishing);
 };
 
-// Frames can be sent upstream using the `_send` method. The frames to be sent are put into the
-// `upstream._queue` first, and are flushed immediately on the beginning of the next turn.
-Stream.prototype._send = function _send(frame) {
-  frame.flags = frame.flags || {};
-  this.upstream._queue.push(frame);
-  if (!this._flushTimer) {
-    this._flushTimer = setImmediate(this._flush.bind(this));
+// The `_receive` method (= `upstream._receive`) gets called when there's an incoming frame.
+Stream.prototype._receive = function _receive(frame, ready) {
+  var callReady = true;
+
+  // * If it's a DATA frame, then push the payload into the output buffer on the other side.
+  //   Call ready when the other side is ready to receive more.
+  if (frame.type === 'DATA') {
+    var moreNeeded = this.push(frame.data);
+    if (!moreNeeded) {
+      this._receiveMore = ready;
+      callReady = false;
+    }
   }
+
+  // * Otherwise it's a control frame. Emit an event to notify interested parties.
+  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._flush = function _flush() {
-  var frame;
-  while(frame = this.upstream._queue.shift()) {
-    this.upstream.emit('sending', frame);
-    this._log.debug({ frame: frame }, 'Sending frame');
-    this.upstream.push(frame);
+// The `_read` method is called when the user side is ready to receive more data. If there's a
+// pending write on the upstream, then call its pending ready callback to receive more frames.
+Stream.prototype._read = function _read() {
+  if (this._receiveMore) {
+    var receiveMore = this._receiveMore;
+    delete this._receiveMore;
+    receiveMore();
   }
-  this._flushTimer = undefined;
 };
 
-// The reason for using an output queue is this. When the stream is finishing (the user calls
-// `end()` on it), then we have to set the `END_STREAM` flag on the last object.
-//
-// If there's no frame in the queue, then we create a 0 length DATA frame. We could do this
-// all the time, but putting the flag on an existing frame is a nice optimization.
-var emptyBuffer = new Buffer(0);
+// The `write` method gets called when there's a write request from the user.
+Stream.prototype._write = function _write(buffer, encoding, ready) {
+  // * Chunking is done by the upstream Flow.
+  var moreNeeded = this.upstream.push({
+    type: 'DATA',
+    data: buffer
+  });
+
+  // * Call ready when upstream is ready to receive more frames.
+  if (moreNeeded) {
+    ready();
+  } else {
+    this._sendMore = ready;
+  }
+};
+
+// The `_send` (= `upstream._send`) method is called when upstream is ready to receive more frames.
+// If there's a pending write on the user side, then call its pending ready callback to receive more
+// writes.
+Stream.prototype._send = function _send() {
+  if (this._sendMore) {
+    var sendMore = this._sendMore;
+    delete this._sendMore;
+    sendMore();
+  }
+};
+
+// When the stream is finishing (the user calls `end()` on it), then we have to set the `END_STREAM`
+// flag on the last frame. If there's no frame in the queue, or if it doesn't support this flag,
+// then we create a 0 length DATA frame. We could do this all the time, but putting the flag on an
+// existing frame is a nice optimization.
+var endFrame = {
+  type: 'DATA',
+  flags: { END_STREAM: true },
+  data: new Buffer(0)
+};
 Stream.prototype._finishing = function _finishing() {
-  var length = this.upstream._queue.length;
-  if (length === 0) {
-    this._send({
-      type: 'DATA',
-      flags: { END_STREAM: true },
-      data: emptyBuffer
-    });
+  delete endFrame.stream;
+  var lastFrame = this.upstream.getLastQueuedFrame();
+  if (lastFrame && ((lastFrame.type === 'DATA') || (lastFrame.type === 'HEADERS'))) {
+    this._log.trace('Marking last frame with END_STREAM flag.');
+    lastFrame.flags.END_STREAM = true;
+    this._transition(true, endFrame);
   } else {
-    var lastFrame = this.upstream._queue[length - 1];
-    lastFrame.flags.END_STREAM = true;
+    this.upstream.push(endFrame);
   }
@@ -188,4 +276,4 @@ };
   this.state = 'IDLE';
-  this.upstream.on('sending', this._transition.bind(this, true));
-  this.upstream.on('receiving', this._transition.bind(this, false));
+  this.on('sending', this._transition.bind(this, true));
+  this.on('receiving', this._transition.bind(this, false));
 };
@@ -233,3 +321,3 @@
         }
-      } else { // TODO: Not well defined. https://github.com/http2/http2-spec/issues/165
+      } else {
         error = 'PROTOCOL_ERROR';
@@ -246,2 +334,3 @@ }
     //   releases the stream reservation.
+    // * An endpoint may receive PRIORITY frame in this state.
     // * An endpoint MUST NOT send any other type of frame in this state.
@@ -253,3 +342,3 @@ case 'RESERVED_LOCAL':
         this._setState('CLOSED');
-      } else { // TODO: Not well defined. https://github.com/http2/http2-spec/issues/165
+      } else if (!(receiving && (frame.type === 'PRIORITY'))) {
         error = 'PROTOCOL_ERROR';
@@ -264,2 +353,3 @@ }
     // * Receiving a HEADERS frame causes the stream to transition to "half closed (local)".
+    // * An endpoint MAY send PRIORITY frames in this state to reprioritize the stream.
     // * Receiving any other type of frame MUST be treated as a stream error of type PROTOCOL_ERROR.
@@ -271,3 +361,3 @@ case 'RESERVED_REMOTE':
         this._setState('HALF_CLOSED_LOCAL');
-      } else {
+      } else if (!(sending && (frame.type === 'PRIORITY'))) {
         error = 'PROTOCOL_ERROR';
@@ -298,6 +388,8 @@ }
     //   flag is received, or when either peer sends a RST_STREAM frame.
+    // * An endpoint MAY send or receive PRIORITY frames in this state to reprioritize the stream.
+    // * WINDOW_UPDATE can be sent by a peer that has sent a frame bearing the END_STREAM flag.
     case 'HALF_CLOSED_LOCAL':
       if ((frame.type === 'RST_STREAM') || (receiving && frame.flags.END_STREAM)) {
         this._setState('CLOSED');
-      } else if (sending) {
+      } else if (sending && !(frame.type === 'PRIORITY') && !(frame.type === 'WINDOW_UPDATE')) {
         error = 'PROTOCOL_ERROR';
@@ -315,6 +407,8 @@ } // Receiving anything is OK
     //   END_STREAM flag, or when either peer sends a RST_STREAM frame.
+    // * An endpoint MAY send or receive PRIORITY frames in this state to reprioritize the stream.
+    // * A receiver MAY receive a WINDOW_UPDATE frame on a "half closed (remote)" stream.
     case 'HALF_CLOSED_REMOTE':
       if ((frame.type === 'RST_STREAM') || (sending && frame.flags.END_STREAM)) {
         this._setState('CLOSED');
-      } else if (receiving) {
+      } else if (receiving && !(frame.type === 'PRIORITY') && !(frame.type === 'WINDOW_UPDATE')) {
         error = 'PROTOCOL_ERROR';
@@ -329,2 +423,8 @@ } // Sending anything is OK
     //   treat that as a stream error of type STREAM_CLOSED.
+    // * WINDOW_UPDATE or PRIORITY frames can be received in this state for a short period after a
+    //   frame containing an END_STREAM flag is sent.  Until the remote peer receives and processes
+    //   the frame bearing the END_STREAM flag, it might send either frame type. Endpoints MUST
+    //   ignore WINDOW_UPDATE frames received in this state, though endpoints MAY choose to treat
+    //   WINDOW_UPDATE frames that arrive a significant time after sending END_STREAM as a
+    //   connection error of type PROTOCOL_ERROR.
     // * If this state is reached as a result of sending a RST_STREAM frame, the peer that receives
@@ -342,3 +442,5 @@ //   the RST_STREAM might have already sent - or enqueued for sending - frames on the stream
         this._setState('RESERVED_REMOTE');
-      } else if (!(sending && (frame.type === 'RST_STREAM'))) {
+      } else if (!(sending && (frame.type === 'RST_STREAM')) &&
+                 !(receiving && (frame.type === 'WINDOW_UPDATE')) &&
+                 !(receiving && (frame.type === 'PRIORITY'))) {
         error = 'PROTOCOL_ERROR';
@@ -376,103 +478,5 @@ } // TODO: act based on the reason for termination.
       this.reset(error);
+      this.emit('error', error);
     }
   }
 };
-
-// [Flow control](http://tools.ietf.org/id/draft-unicorn-httpbis-http2-01.html#rfc.section.6.9)
-// --------------
-
-// Flow control in HTTP/2.0 is implemented using a window kept by each sender on every stream.
-// The flow control window is a simple integer value that indicates how many bytes of data the
-// sender is permitted to transmit. Two flow control windows are applicable; the stream flow control
-// window and the connection flow control window. The stream only manages the flow control `window`.
-Stream.prototype._initializeFlowControl = function _initializeFlowControl() {
-  this._read = utils.noop;
-  this.upstream.on('receiving', this._receiveData.bind(this));
-
-  this._window = INITIAL_WINDOW_SIZE;
-  this.upstream.on('receiving', this._updateWindow.bind(this));
-};
-
-// When a HTTP/2.0 connection is first established, new streams are created with an initial flow
-// control window size of 65535 bytes.
-var INITIAL_WINDOW_SIZE = 65535;
-
-// A SETTINGS frame can alter the initial flow control window size for all current streams. When the
-// value of SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST adjust the size of all stream by
-// calling the `setInitialWindowSize` method. The window size has to be modified by the difference
-// between the new value and the old value.
-Stream.prototype.setInitialWindowSize = function setInitialWindowSize(initialWindowSize) {
-  this._window = this._window - this._initialWindowSize + initialWindowSize;
-  this._initialWindowSize = initialWindowSize;
-};
-
-// Flow control can be disabled for all streams on the connection using the `disableFlowControl`
-// method. This may happen when there's a SETTINGS frame received with the
-// SETTINGS_FLOW_CONTROL_OPTIONS setting.
-Stream.prototype.disableFlowControl = function disableFlowControl() {
-  this._window = Infinity;
-};
-
-// The `_updateWindow` method gets called every time there's an incoming frame. It filters out
-// WINDOW_UPDATE frames, and then modifies the modifies the flow control window:
-//
-// * Flow control can be disabled for an individual stream by sending a WINDOW_UPDATE with the
-//   END_FLOW_CONTROL flag set. The payload of a WINDOW_UPDATE frame that has the END_FLOW_CONTROL
-//   flag set is ignored.
-// * A sender that receives a WINDOW_UPDATE frame updates the corresponding window by the amount
-//   specified in the frame.
-Stream.prototype._updateWindow = function _updateWindow(frame) {
-  if (frame.type === 'WINDOW_UPDATE') {
-    if (frame.flags.END_FLOW_CONTROL) {
-      this.disableFlowControl();
-    } else {
-      this._window += frame.window_size;
-    }
-    this.emit('window_update');
-  }
-};
-
-// When the user wants to write a buffer into the stream
-Stream.prototype._write = function _write(buffer, encoding, done) {
-  // * 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);
-  var sent = 0;
-
-  // * 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
-  while ((chunks.length > 0) && (chunks[0].length <= this._window)) {
-    var chunk = chunks.shift();
-    sent += chunk.length;
-    this._send({
-      type: 'DATA',
-      flags: {},
-      data: chunk
-    });
-
-    // * After sending a flow controlled frame, the sender reduces the space available the window by
-    //   the length of the transmitted frame. For flow control calculations, the 8 byte frame header
-    //   is not counted.
-    this._window -= chunk.length;
-  }
-
-  // * If all of the chunks are sent, we are done
-  if (chunks.length === 0) {
-    done();
-  }
-
-  // * Otherwise the process has to continue when a window_update occurs. It is guaranteed by
-  //   the Duplex stream class, that there will be no more calls to `_write` until we are done
-  else {
-    this.once('window_update', this._write.bind(this, buffer.slice(sent), encoding, done));
-  }
-};
-
-Stream.prototype._receiveData = function _receiveData(frame) {
-  if (frame.type === 'DATA') {
-    this.push(frame.data);
-  }
-
-  if (frame.flags.END_STREAM) {
-    this.push(null);
-  }
-};

package.json

 {
   "name": "http2",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "description": "An HTTP/2 server implementation",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

README.md

@@ -11,7 +11,8 @@ node-http2
 
-I post weekly status updates [on my blog][2]. Short version: the first version of the public API is
-in place. NPN negotiation works (no ALPN or Upgrade mechanism yet). Main missing items will be
-tracked in the issue tracker.
+I post weekly status updates [on my blog][1]. Short version: the first version of the public API is
+in place, server push is not exposed yet, prioritization and ALPN support is not yet done. Main
+missing items will be tracked in the issue tracker under the label [feature][2].
 
-[2]: http://gabor.molnar.es/blog/categories/google-summer-of-code/
+[1]: http://gabor.molnar.es/blog/categories/google-summer-of-code/
+[2]: https://github.com/molnarg/node-http2/issues?labels=feature&state=open
 
@@ -48,3 +49,3 @@ Installation
 
-http2.http.createServer(options, function(request, response) {
+http2.createServer(options, function(request, response) {
   response.end('Hello world!');
@@ -99,13 +100,13 @@ }).listen(8080);
 
-* [mocha][3] for tests
-* [chai][4] for assertions
-* [istanbul][5] for code coverage analysis
-* [docco][6] for developer documentation
-* [bunyan][7] for logging
+* [mocha][1] for tests
+* [chai][2] for assertions
+* [istanbul][3] for code coverage analysis
+* [docco][4] for developer documentation
+* [bunyan][5] for logging
 
-[3]: http://visionmedia.github.io/mocha/
-[4]: http://chaijs.com/
-[5]: https://github.com/gotwarlost/istanbul
-[6]: http://jashkenas.github.io/docco/
-[7]: https://github.com/trentm/node-bunyan
+[1]: http://visionmedia.github.io/mocha/
+[2]: http://chaijs.com/
+[3]: https://github.com/gotwarlost/istanbul
+[4]: http://jashkenas.github.io/docco/
+[5]: https://github.com/trentm/node-bunyan
 
@@ -134,5 +135,5 @@ ### Developer documentation ###
 
-There's a hosted version of the detailed (line-by-line) coverage report [here][8].
+There's a hosted version of the detailed (line-by-line) coverage report [here][1].
 
-[8]: http://molnarg.github.io/node-http2/coverage/lcov-report/lib/
+[1]: http://molnarg.github.io/node-http2/coverage/lcov-report/lib/
 
@@ -139,0 +140,0 @@ ### Logging ###

test/compressor.js

 var expect = require('chai').expect;
-var concat = require('../lib/utils').concat;
 
@@ -9,2 +8,17 @@ var compressor = require('../lib/compressor');
 
+// Concatenate an array of buffers into a new buffer
+function concat(buffers) {
+  var size = 0;
+  for (var i = 0; i < buffers.length; i++) {
+    size += buffers[i].length;
+  }
+
+  var concatenated = new Buffer(size);
+  for (var cursor = 0, j = 0; j < buffers.length; cursor += buffers[j].length, j++) {
+    buffers[j].copy(concatenated, cursor);
+  }
+
+  return concatenated;
+}
+
 var test_integers = [{
@@ -11,0 +25,0 @@ N: 5,

test/framer.js

 var expect = require('chai').expect;
-var concat = require('../lib/utils').concat;
 
@@ -8,2 +7,17 @@ var framer = require('../lib/framer');
 
+// Concatenate an array of buffers into a new buffer
+function concat(buffers) {
+  var size = 0;
+  for (var i = 0; i < buffers.length; i++) {
+    size += buffers[i].length;
+  }
+
+  var concatenated = new Buffer(size);
+  for (var cursor = 0, j = 0; j < buffers.length; cursor += buffers[j].length, j++) {
+    buffers[j].copy(concatenated, cursor);
+  }
+
+  return concatenated;
+}
+
 var frame_types = {
@@ -10,0 +24,0 @@ DATA:          ['data'],

test/stream.js

@@ -12,7 +12,14 @@ var expect = require('chai').expect;
     }
-  }
+  };
 }
 
+function createStream() {
+  var stream = new Stream();
+  stream.upstream._window = Infinity;
+  stream.upstream._remoteFlowControlDisabled = true;
+  return stream;
+}
+
 // Execute a list of commands and assertions
-var recorded_events = ['state', 'error', 'window_update', 'headers', 'promise']
+var recorded_events = ['state', 'error', 'window_update', 'headers', 'promise'];
 function execute_sequence(stream, sequence, done) {
@@ -22,7 +29,6 @@ if (!done) {
     sequence = stream;
-    stream = new Stream();
+    stream = createStream();
   }
 
   var outgoing_frames = [];
-  stream.upstream.on('sending', outgoing_frames.push.bind(outgoing_frames));
 
@@ -39,5 +45,7 @@ var emit = stream.emit, events = [];
   sequence.forEach(function(step) {
-    if ('method' in step || 'incoming' in step || 'wait' in step || 'set_state' in step) {
+    if ('method' in step || 'incoming' in step || 'outgoing' in step || 'wait' in step || 'set_state' in step) {
       commands.push(step);
-    } else {
+    }
+
+    if ('outgoing' in step || 'event' in step) {
       checks.push(step);
@@ -56,2 +64,5 @@ }
         execute(callback);
+      } else if ('outgoing' in command) {
+        outgoing_frames.push(stream.upstream.read());
+        execute(callback);
       } else if ('set_state' in command) {
@@ -96,3 +107,2 @@ stream.state = command.set_state;
     { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
-    { type: 'PRIORITY', flags: {}, priority: 1 },
     { type: 'PUSH_PROMISE', flags: {}, headers: {} },
@@ -114,5 +124,3 @@ { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
     { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
-    { type: 'PRIORITY', flags: {}, priority: 1 },
-    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
-    { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
+    { type: 'PUSH_PROMISE', flags: {}, headers: {} }
   ],
@@ -127,3 +135,3 @@ CLOSED: [ // TODO
       Object.keys(invalid_frames).forEach(function(state) {
-        it('should answer RST_STREAM for invalid incoming frames in ' + state + ' state', function(done) {
+        it('should emit error, and answer RST_STREAM for invalid incoming frames in ' + state + ' state', function(done) {
           var left = invalid_frames[state].length + 1;
@@ -139,3 +147,8 @@ function one_done() {
           invalid_frames[state].forEach(function(invalid_frame) {
-            execute_sequence([
+            var stream = createStream();
+            var error_emitted = false;
+            stream.on('error', function() {
+              error_emitted = true;
+            });
+            execute_sequence(stream, [
               { set_state: state },
@@ -145,3 +158,6 @@ { incoming : invalid_frame },
               { outgoing : { type: 'RST_STREAM', flags: {}, error: 'PROTOCOL_ERROR' } }
-            ], one_done);
+            ], function sequence_ready() {
+              expect(error_emitted).to.equal(true);
+              one_done();
+            });
           });
@@ -178,4 +194,4 @@ });
           { incoming: { type: 'HEADERS', flags: { }, headers: { ':path': '/' } } },
+          { event   : { name: 'state', data: 'OPEN' } },
           { event   : { name: 'headers', data: { ':path': '/' } } },
-          { event   : { name: 'state', data: 'OPEN' } },
 
@@ -201,4 +217,4 @@ { wait    : 5 },
         var payload = new Buffer(5);
-        var original_stream = new Stream();
-        var promised_stream = new Stream();
+        var original_stream = createStream();
+        var promised_stream = createStream();
 
@@ -210,5 +226,5 @@ done = callNTimes(2, done);
           { incoming: { type: 'HEADERS', flags: { END_STREAM: true }, headers: { ':path': '/' } } },
-          { event   : { name: 'headers', data: { ':path': '/' } } },
           { event   : { name: 'state', data: 'OPEN' } },
           { event   : { name: 'state', data: 'HALF_CLOSED_REMOTE' } },
+          { event   : { name: 'headers', data: { ':path': '/' } } },
 
@@ -250,4 +266,4 @@ // sending response headers
         var payload = new Buffer(5);
-        var original_stream = new Stream();
-        var promised_stream = new Stream();
+        var original_stream = createStream();
+        var promised_stream = createStream();
 
@@ -285,4 +301,4 @@ done = callNTimes(2, done);
           { incoming: { type: 'HEADERS', flags: { END_STREAM: false }, headers: { ':status': 200 } } },
+          { event   : { name: 'state', data: 'HALF_CLOSED_LOCAL' } },
           { event   : { name: 'headers', data: { ':status': 200 } } },
-          { event   : { name: 'state', data: 'HALF_CLOSED_LOCAL' } },
 
@@ -289,0 +305,0 @@ // push data

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

854104

increased by4.63%

Lines of code

4426

increased by6.78%

Number of lines in readme file

160

increased by0.63%

Number of medium supply chain risk alerts

4

decreased by-20%

Worsened metrics

Number of package files

43

decreased by-2.27%

Number of low supply chain risk alerts

7

increased by16.67%

No dependency changes