issue #7 - first pass on mongo-style filter conditions

README.md

@@ -141,6 +141,21 @@ The Array-like API also includes `every`, `some` and `forEach`. On the other han
 
 The `forEach`, `every` and `some` functions are reducers and take a continuation callback, like `reduce` (see example further down).
 
+Note: the `filter`, `every` and `some` methods can also be controlled by a mongodb filter condition rather than a function. The following are equivalent:
+
+``` javascript
+// filter expressed as a function
+reader =  numberReader(1000).filter(function(_, n) {
+	return n >= 10 && n < 20;
+});
+
+// mongo-style filter
+reader =  numberReader(1000).filter({
+	$gte: 10,
+	$lt: 20,
+});
+```
+
 <a name="pipe"/>
 ## Pipe
 
@@ -197,6 +212,8 @@ infiniteReader().until(function(_, n) {
 }).pipe(_, ez.devices.console.log);
 ```
 
+Note: `while` and `until` conditions can also be expressed as mongodb conditions.
+
 <a name="transforms"/>
 ## Transformations
 

lib/predicate._js

@@ -0,0 +1,137 @@
+"use strict";
+
+function pfalse(_, obj) {
+	return false;
+}
+
+function ptrue(_, obj) {
+	return true;
+}
+
+var ops = {
+	$eq: function(val) {
+		return function(_, v) {
+			return v == val;
+		};
+	},
+	$ne: function(val) {
+		return function(_, v) {
+			return v != val;
+		};
+	},
+	$gt: function(val) {
+		return function(_, v) {
+			return v > val;
+		};
+	},
+	$gte: function(val) {
+		return function(_, v) {
+			return v >= val;
+		};
+	},
+	$lt: function(val) {
+		return function(_, v) {
+			return v < val;
+		};
+	},
+	$lte: function(val) {
+		return function(_, v) {
+			return v <= val;
+		};
+	},
+	$in: function(val) {
+		return function(_, v) {
+			return val.indexOf(v) >= 0;
+		}
+	},
+	$nin: function(val) {
+		return function(_, v) {
+			return val.indexOf(v) < 0;
+		}
+	},
+	$and: function(val) {
+		return and(val.map(exports.predicate));
+	},
+	$or: function(val) {
+		return or(val.map(exports.predicate));
+	},
+	$nor: function(val) {
+		return not(or(val.map(exports.predicate)));
+	},
+	$not: function(val) {
+		return not(exports.predicate(val));
+	},
+}
+
+function re_test(re) {
+	return function(_, val) {
+		return re.test(val);
+	}
+}
+
+function not(predicate) {
+	return function(_, obj) {
+		return !predicate(_, obj);
+	}
+}
+
+function or(predicates) {
+	if (predicates.length === 0) return pfalse;
+	if (predicates.length === 1) return predicates[0];
+	return function(_, obj) {
+		return predicates.some_(_, function(_, predicate) {
+			return predicate(_, obj);
+		});
+	}
+}
+
+function and(predicates) {
+	if (predicates.length === 0) return ptrue;
+	if (predicates.length === 1) return predicates[0];
+	return function(_, obj) {
+		return predicates.every_(_, function(_, predicate) {
+			return predicate(_, obj);
+		});
+	}
+}
+
+function compose(f, g) {
+	return function(_, obj) {
+		return f(_, g(_, obj));
+	}
+}
+
+function deref(key) {
+	return function(_, obj) {
+		if (obj == null) return undefined;
+		var v = obj[key];
+		return typeof v === "function" ? v(_) : v;
+	}
+}
+
+function walk(p) {
+	var i = p.indexOf('.');
+	if (i >= 0) {
+		return compose(walk(p.substring(i + 1)), walk(p.substring(0, i)));
+	} else {
+		return deref(p);
+	}
+}
+
+exports.predicate = function(val) {
+	if (val instanceof RegExp) {
+		return re_test(val);
+	} else if (typeof val === "object" && val) {
+		return and(Object.keys(val).map(function(k) {
+			var v = val[k];
+			if (k[0] === '$') {
+				if (!ops[k]) throw new Error("bad operator: " + k);
+				return ops[k](v);
+			} else {
+				return compose(exports.predicate(v), walk(k));
+			}
+		}));
+	} else {
+		return ops.$eq(val);
+	}
+};

lib/reader._js

@@ -30,6 +30,8 @@
 /// 
 var streams = require('streamline-streams/lib/streams');
 var flows = require('streamline/lib/util/flows');
+var predicate = require('./predicate').predicate;
+
 var generic;
 
 var Decorated = function Decorated(read) {
@@ -79,6 +81,7 @@ exports.decorate = function(proto) {
 	///   Stops streaming and returns false as soon as `fn` returns false on an entry.
 	proto.every = function(_, fn, thisObj) {
 		thisObj = thisObj !== undefined ? thisObj : this;
+		if (typeof fn !== 'function') fn = predicate(fn);
 		var self = this;
 		while (true) {
 			var val = self.read(_);
@@ -94,6 +97,7 @@ exports.decorate = function(proto) {
 	///   Stops streaming and returns true as soon as `fn` returns true on an entry.
 	proto.some = function(_, fn, thisObj) {
 		thisObj = thisObj !== undefined ? thisObj : this;
+		if (typeof fn !== 'function') fn = predicate(fn);
 		var self = this;
 		while (true) {
 			var val = self.read(_);
@@ -166,6 +170,7 @@ exports.decorate = function(proto) {
 	///   Returns another stream on which other operations may be chained.
 	proto.filter = function(fn, thisObj) {
 		thisObj = thisObj !== undefined ? thisObj : this;
+		if (typeof fn !== 'function') fn = predicate(fn);
 		return this.transform(function(_, reader, writer) {
 			for (var i = 0, val;
 			(val = reader.read(_)) !== undefined; i++) {
@@ -180,6 +185,7 @@ exports.decorate = function(proto) {
 	///   Returns another stream on which other operations may be chained.
 	proto.until = function(fn, thisObj) {
 		thisObj = thisObj !== undefined ? thisObj : this;
+		if (typeof fn !== 'function') fn = predicate(fn);
 		return this.transform(function(_, reader, writer) {
 			for (var i = 0, val;
 			(val = reader.read(_)) !== undefined; i++) {
@@ -197,6 +203,7 @@ exports.decorate = function(proto) {
 	///   Returns another stream on which other operations may be chained.
 	proto.
 	while = function(fn, thisObj) {
+		if (typeof fn !== 'function') fn = predicate(fn);
 		return this.until(function(_, val, i) {
 			return !fn.call(thisObj, _, val, i);
 		}, thisObj);