pullstream - npm Package Compare versions

Comparing version 0.0.2 to 0.0.3

package.json

 {
   "name": "pullstream",
-  "version": "0.0.2",
+  "version": "0.0.3",
   "description": "A stream you can pull data from.",
@@ -18,3 +18,4 @@ "main": "pullstream.js",
   "keywords": [
-    "stream", "pull"
+    "stream",
+    "pull"
   ],
@@ -21,0 +22,0 @@ "devDependencies": {

pullstream.js

@@ -10,5 +10,8 @@ 'use strict';
 
-function PullStream() {
+function PullStream(opts) {
   var self = this;
   Stream.apply(this);
+  this.opts = opts || {};
+  this.opts.minBufferSize = this.opts.minBufferSize | (1 * 1024 * 1024);
+  this.opts.maxBufferSize = this.opts.maxBufferSize | (10 * 1024 * 1024);
   this.readable = false;
@@ -35,2 +38,5 @@ this.writable = true;
   this._buffer.write(data);
+  if (this._buffer.maxSize() > this.opts.maxBufferSize && this._srcStream) {
+    this._srcStream.pause();
+  }
   this.process();
@@ -67,4 +73,8 @@ return true;
 PullStream.prototype._finish = function () {
-  this.emit('end');
-  this.emit('close');
+  var self = this;
+  process.nextTick(function () {
+    self.emit('end');
+    self.emit('close');
+  });
+  this._finish = function () {};
 };
@@ -90,12 +100,17 @@
         var results = self._buffer.getContents(len);
+        self._resumeSrcStream();
         results.posInStream = self._positionInStream;
         self._positionInStream += results.length;
-        callback(null, results);
+        process.nextTick(function () {
+          callback(null, results);
 
-        if (self._recvEnd && self._buffer.size() === 0) {
-          self._finish();
-        }
+          if (self._recvEnd && self._buffer.size() === 0) {
+            self._finish();
+          }
+        });
       } else if (self._recvEnd && self._buffer.size() === 0) {
         callback(new Error('End of Stream'));
         self._finish();
+      } else {
+        self._resumeSrcStream();
       }
@@ -130,2 +145,3 @@ }
         var results = self._buffer.getContents(lenToRemove);
+        self._resumeSrcStream();
         results.posInStream = self._positionInStream;
@@ -144,2 +160,4 @@ self._positionInStream += results.length;
         }
+      } else {
+        self._resumeSrcStream();
       }
@@ -160,2 +178,8 @@
 
+PullStream.prototype._resumeSrcStream = function () {
+  if (this._srcStream && this._buffer.size() < this.opts.minBufferSize) {
+    this._srcStream.resume();
+  }
+};
+
 PullStream.prototype.pause = function () {
@@ -162,0 +186,0 @@ this.paused = true;

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

20393

increased by3.94%

Lines of code

489

increased by4.94%

No dependency changes