understream - npm Package Compare versions

Comparing version 0.8.0 to 0.8.1

lib-js/readables/file.js

@@ -19,3 +19,5 @@ // Generated by CoffeeScript 1.6.3
     this.stream_opts = stream_opts;
-    File.__super__.constructor.call(this, this.stream_opts);
+    File.__super__.constructor.call(this, _(this.stream_opts).extend({
+      objectMode: false
+    }));
     this.wrap(fs.createReadStream(path, options));
@@ -22,0 +24,0 @@ }

lib-js/transforms/batch.js

@@ -42,4 +42,8 @@ // Generated by CoffeeScript 1.6.3
 
+  Batch.prototype._extra_report_string = function() {
+    return "" + this._buffer.length;
+  };
+
   return Batch;
 
 })(Transform);

lib-js/transforms/process.js

@@ -31,4 +31,8 @@ // Generated by CoffeeScript 1.6.3
 
+  Process.prototype._extra_report_string = function() {
+    return ("stdin:" + this.process.stdin._writableState.length) + (" pid:" + this.process.pid) + (" stdout:" + this.process.stdout._readableState.length);
+  };
+
   return Process;
 
 })(Duplex);

lib-js/transforms/queue.js

@@ -34,3 +34,3 @@ // Generated by CoffeeScript 1.6.3
     _(this.options).defaults({
-      concurrency: 1000
+      concurrency: this.stream_opts.highWaterMark || 10
     });
@@ -37,0 +37,0 @@ this.q = async.queue(function(payload, cb) {

lib-js/transforms/split.js

@@ -23,2 +23,3 @@ // Generated by CoffeeScript 1.6.3
     Split.__super__.constructor.call(this, this.stream_opts);
+    this._readableState.objectMode = true;
     if ((this.options == null) || !(_(this.options).isString() || this.options instanceof RegExp || _(this.options).isObject())) {
@@ -25,0 +26,0 @@ throw new Error("Split requires separator");

lib-js/understream.js

 // Generated by CoffeeScript 1.6.3
-var ArrayStream, DevNull, EventEmitter, PassThrough, Readable, StreamCombiner, Understream, Writable, debug, domain, fs, is_readable, pipe_streams_together, _, _ref,
+var ArrayStream, DevNull, EventEmitter, PassThrough, Readable, StreamCombiner, Understream, Writable, add_reporter, debug, domain, fs, is_readable, pipe_streams_together, state_to_string, to_report_string, _, _ref,
   __slice = [].slice,
@@ -30,2 +30,30 @@ __bind = function(fn, me){ return function(){ return fn.apply(me, arguments); }; },
 
+state_to_string = function(state) {
+  if (state != null) {
+    return (state.length || '') + (state.objectMode ? 'o' : 'b');
+  } else {
+    return '';
+  }
+};
+
+to_report_string = function(stream) {
+  return _([state_to_string(stream._writableState), stream.constructor.name, typeof stream._extra_report_string === "function" ? stream._extra_report_string() : void 0, state_to_string(stream._readableState)]).compact().join(' ');
+};
+
+add_reporter = function(streams) {
+  var interval, report;
+  report = function() {
+    return console.log(_(streams).map(to_report_string).join(' | '));
+  };
+  interval = setInterval(report, 5000);
+  _(streams).each(function(stream) {
+    return stream.on('error', function() {
+      return clearInterval(interval);
+    });
+  });
+  return _(streams).last().on('finish', function() {
+    return clearInterval(interval);
+  });
+};
+
 pipe_streams_together = function() {
@@ -141,3 +169,3 @@ var i, streams, _i, _ref1, _results;
   Understream.prototype.run = function(cb) {
-    var batch_stream, dmn, handler, interval, report, result, stream, _i, _len, _ref1,
+    var batch_stream, dmn, handler, result,
       _this = this;
@@ -147,12 +175,2 @@ if (!_(cb).isFunction()) {
     }
-    report = function() {
-      var str;
-      str = '';
-      _(_this._streams).each(function(stream) {
-        var _ref1, _ref2;
-        return str += "" + stream.constructor.name + "(" + (((_ref1 = stream._writableState) != null ? _ref1.length : void 0) || '') + " " + (((_ref2 = stream._readableState) != null ? _ref2.length : void 0) || '') + ") ";
-      });
-      return console.log(str);
-    };
-    interval = setInterval(report, 5000);
     if (cb.length === 2) {
@@ -169,5 +187,6 @@ result = [];
     }
+    add_reporter(this._streams);
     dmn = domain.create();
     handler = function(err) {
-      clearInterval(interval);
+      dmn.dispose();
       if (cb.length === 1) {
@@ -180,8 +199,6 @@ return cb(err);
     _(this._streams).last().on('finish', handler);
+    _(this._streams).each(function(stream) {
+      return stream.on('error', handler);
+    });
     dmn.on('error', handler);
-    _ref1 = this._streams;
-    for (_i = 0, _len = _ref1.length; _i < _len; _i++) {
-      stream = _ref1[_i];
-      dmn.add(stream);
-    }
     dmn.run(function() {
@@ -195,2 +212,3 @@ debug('running');
   Understream.prototype.readable = function() {
+    add_reporter(this._streams);
     pipe_streams_together.apply(null, this._streams);
@@ -201,2 +219,3 @@ return this._streams[this._streams.length - 1];
   Understream.prototype.duplex = function() {
+    add_reporter(this._streams);
     return (function(func, args, ctor) {
@@ -203,0 +222,0 @@ ctor.prototype = func.prototype;

package.json

 {
   "name": "understream",
-  "version": "0.8.0",
+  "version": "0.8.1",
   "description": "stream helpers",
@@ -5,0 +5,0 @@ "engines": {

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

108880

increased by1.31%

Lines of code

1197

increased by2.13%

No dependency changes