import Transform from './Transform';
import {inherits} from '../util/Functions';

/**
 * Filters data tuples according to a predicate function.
 * @constructor
 * @param {object} params - The parameters for this operator.
 * @param {function(object): *} params.test - The predicate function that
 *   determines a tuple's filter status. Truthy values pass the filter.
 */
export default function Filter(params) {
  Transform.call(this, {}, params);
}

var prototype = inherits(Filter, Transform);

prototype.transform = function(_, pulse) {
  var test = _.test,
      cache = this.value, // cache ids of filtered tuples
      output = pulse.fork(),
      flags = pulse.MOD | (_.modified('test') ? pulse.REFLOW : 0);

  pulse.visit(pulse.REM, function(x) {
    if (cache[x._id] !== 1) output.rem.push(x);
    else cache[x._id] = 0;
  });

  pulse.visit(pulse.ADD, function(x) {
    if (test(x)) output.add.push(x);
    else cache[x._id] = 1;
  });

  pulse.visit(flags, function(x) {
    var b = test(x),
        s = (cache[x._id] === 1);
    if (b && s) {
      cache[x._id] = 0;
      output.add.push(x);
    } else if (b && !s) {
      output.mod.push(x);
    } else if (!b && s) {
      // do nothing, keep cache true
    } else { // !b && !s
      output.rem.push(x);
      cache[x._id] = 1;
    }
  });

  return output;
};

@@ -652,67 +652,86 @@
     
-flags = pulse.MOD %7C (_.modified('test') ? pulse.REFLOW : 0)
+add = output.add,%0A      rem = output.rem,%0A      mod = output.mod, isMod = true
 ;%0A%0A 
@@ -776,16 +776,17 @@
     if (
+!
 cache%5Bx.
@@ -793,23 +793,10 @@
 _id%5D
+)
  
-!== 1) output.
 rem.
@@ -894,23 +894,16 @@
 est(x)) 
-output.
 add.push
@@ -947,35 +947,16 @@
 %0A%0A  
-pulse.visit(flags, 
 function
 (x) 
@@ -943,32 +943,40 @@
  %7D);%0A%0A  function
+ revisit
 (x) %7B%0A    var b 
@@ -998,17 +998,16 @@
     s = 
-(
 cache%5Bx.
@@ -1014,15 +1014,8 @@
 _id%5D
- === 1)
 ;%0A  
@@ -1060,23 +1060,16 @@
 ;%0A      
-output.
 add.push
@@ -1084,24 +1084,25 @@
  %7D else if (
+!
 b && !s) %7B%0A 
@@ -1106,26 +1106,43 @@
 %7B%0A      
-output.mod
+cache%5Bx._id%5D = 1;%0A      rem
 .push(x)
@@ -1158,22 +1158,31 @@
 lse if (
-!
+isMod && 
 b && 
+!
 s) %7B%0A   
@@ -1188,125 +1188,158 @@
     
-// do nothing, keep cache true%0A    %7D else %7B // !b && !s%0A      output.rem.push(x
+mod.push(x);%0A    %7D%0A  %7D%0A%0A  pulse.visit(pulse.MOD, revisit
 );%0A
+%0A
   
-    cache%5Bx._id%5D = 1;%0A    %7D
+if (_.modified('test')) %7B%0A    isMod = false;%0A    pulse.visit(pulse.REFLOW, revisit);
 %0A  %7D
-);
 %0A%0A  

Make filter reflow more efficient.