understream - npm Package Compare versions

Comparing version 0.7.8 to 0.7.9

lib-js/transforms/join.js

 // Generated by CoffeeScript 1.6.3
-var HashAccumulator, HashJoin, Join, Readable, SortedMergeJoin, Transform, Writable, debug, hash_fn, nextTick, util, _, _ref,
+var HashAccumulator, HashJoin, Join, Readable, SortedMergeJoin, Transform, Writable, async, debug, hash_fn, util, _, _ref,
   __bind = function(fn, me){ return function(){ return fn.apply(me, arguments); }; },
@@ -15,3 +15,3 @@ __hasProp = {}.hasOwnProperty,
 
-nextTick = require('../nextTick');
+async = require('async');
 
@@ -155,4 +155,10 @@ hash_fn = function(source, onn) {
 SortedMergeJoin = (function(_super) {
+  var ended;
+
   __extends(SortedMergeJoin, _super);
 
+  ended = function(stream) {
+    return stream._readableState.ended && _.isEmpty(stream._readableState.buffer);
+  };
+
   function SortedMergeJoin(stream_opts, _arg) {
@@ -164,29 +170,46 @@ this.right = _arg.right, this.key = _arg.key;
   SortedMergeJoin.prototype._transform = function(chunk, enc, cb) {
-    var l, next, r;
-    next = function() {
-      return nextTick(cb);
-    };
+    var l,
+      _this = this;
     l = chunk;
-    while ((r = this.right.read()) != null) {
-      if (l[this.key] > r[this.key]) {
-        this.push(r);
-      } else if (l[this.key] < r[this.key]) {
-        this.push(l);
-        this.right.unshift(r);
-        return next();
-      } else if (l[this.key] === r[this.key]) {
-        this.push(_.extend({}, l, r));
-        return next();
+    return async.whilst((function() {
+      return (l != null) && !ended(_this.right);
+    }), function(cb_w) {
+      var r;
+      r = _this.right.read();
+      if (r != null) {
+        switch (false) {
+          case !(l[_this.key] > r[_this.key]):
+            _this.push(r);
+            break;
+          case !(l[_this.key] < r[_this.key]):
+            _this.push(l);
+            _this.right.unshift(r);
+            l = null;
+            break;
+          case l[_this.key] !== r[_this.key]:
+            _this.push(_.extend({}, l, r));
+            l = null;
+        }
       }
-    }
-    this.push(l);
-    return next();
+      return setImmediate(cb_w);
+    }, function() {
+      if (l != null) {
+        _this.push(l);
+      }
+      return cb();
+    });
   };
 
   SortedMergeJoin.prototype._flush = function(cb) {
-    var r;
-    while ((r = this.right.read()) != null) {
-      this.push(r);
-    }
-    return nextTick(cb);
+    var _this = this;
+    return async.whilst((function() {
+      return !ended(_this.right);
+    }), function(cb_w) {
+      var r;
+      r = _this.right.read();
+      if (r != null) {
+        _this.push(r);
+      }
+      return setImmediate(cb_w);
+    }, cb);
   };
@@ -193,0 +216,0 @@

package.json

 {
   "name": "understream",
-  "version": "0.7.8",
+  "version": "0.7.9",
   "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

107589

increased by2.55%

Lines of code

1172

increased by1.56%

No dependency changes