Merge pull request #222 from simllll/master

memory leak fix (by argon) and another 'upstream is not defined' check

Author: molnarg

lib/protocol/connection.js

@@ -214,15 +214,18 @@ Connection.prototype._insert = function _insert(stream, priority) {
 };
 
 Connection.prototype._reprioritize = function _reprioritize(stream, priority) {
+  this._removePrioritisedStream(stream);
+  this._insert(stream, priority);
+};
+
+Connection.prototype._removePrioritisedStream = function _removePrioritisedStream(stream) {
   var bucket = this._streamPriorities[stream._priority];
   var index = bucket.indexOf(stream);
   assert(index !== -1);
   bucket.splice(index, 1);
   if (bucket.length === 0) {
     delete this._streamPriorities[stream._priority];
   }
-
-  this._insert(stream, priority);
 };
 
 // Creating an *inbound* stream with the given ID. It is called when there's an incoming frame to
@@ -246,9 +249,18 @@ Connection.prototype.createStream = function createStream() {
   var stream = new Stream(this._log, this);
   this._allocatePriority(stream);
 
+  stream.on('end', this._removeStream.bind(this, stream));
+
   return stream;
 };
 
+Connection.prototype._removeStream = function _removeStream(stream) {
+  this._log.trace('Removing outbound stream.');
+
+  delete this._streamIds[stream.id];
+  this._removePrioritisedStream(stream);
+};
+
 // Multiplexing
 // ------------
 
@@ -290,7 +302,7 @@ priority_loop:
     //   2. if there's no frame, skip this stream
     //   3. if forwarding this frame would make `streamCount` greater than `streamLimit`, skip
     //      this stream
-    //   4. adding stream to the bucket of the next round
+    //   4. adding stream to the bucket of the next round unless it has ended
     //   5. assigning an ID to the frame (allocating an ID to the stream if there isn't already)
     //   6. if forwarding a PUSH_PROMISE, allocate ID to the promised stream
     //   7. forwarding the frame, changing `streamCount` as appropriate
@@ -299,6 +311,7 @@ priority_loop:
     while (bucket.length > 0) {
       for (var index = 0; index < bucket.length; index++) {
         var stream = bucket[index];
+        if(!stream || !stream.upstream) continue;
         var frame = stream.upstream.read((this._window > 0) ? this._window : -1);
 
         if (!frame) {
@@ -308,7 +321,11 @@ priority_loop:
           continue;
         }
 
-        nextBucket.push(stream);
+        if (!stream._ended) {
+          nextBucket.push(stream);
+        } else {
+          delete this._streamIds[stream.id];
+        }
 
         if (frame.stream === undefined) {
           frame.stream = stream.id || this._allocateId(stream);