diff --git a/README.md b/README.md
index b1e4f83..0e34a51 100644
--- a/README.md
+++ b/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
 
diff --git a/lib/predicate._js b/lib/predicate._js
new file mode 100644
index 0000000..cdeb671
--- /dev/null
+++ b/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);
+	}
+};
\ No newline at end of file
diff --git a/lib/reader._js b/lib/reader._js
index fa86891..62fefb6 100644
--- a/lib/reader._js
+++ b/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);
diff --git a/test/common/predicate-test._js b/test/common/predicate-test._js
new file mode 100644
index 0000000..f72dd76
--- /dev/null
+++ b/test/common/predicate-test._js
@@ -0,0 +1,317 @@
+"use strict";
+QUnit.module(module.id);
+
+var predicate = require("ez-streams/lib/predicate").predicate;
+
+function t(_, pred, obj, result) {
+	equals(predicate(pred)(_, obj), result, JSON.stringify(pred) + " with " + JSON.stringify(obj) + " => " + result);
+}
+
+asyncTest("direct values", 6, function(_) {
+	t(_, 5, 5, true);
+	t(_, 5, 6, false);
+	t(_, 'a', 'a', true);
+	t(_, 'a', 'aa', false);
+	t(_, true, true, true);
+	t(_, true, false, false);
+	start();
+});
+
+asyncTest("gt", 3, function(_) {
+	t(_, {
+		$gt: 4,
+	}, 5, true);
+
+	t(_, {
+		$gt: 5,
+	}, 5, false);
+
+	t(_, {
+		$gt: 6,
+	}, 5, false);
+
+	start();
+});
+
+asyncTest("gte", 3, function(_) {
+	t(_, {
+		$gte: 4,
+	}, 5, true);
+
+	t(_, {
+		$gte: 5,
+	}, 5, true);
+
+	t(_, {
+		$gte: 6,
+	}, 5, false);
+
+	start();
+});
+
+asyncTest("lt", 3, function(_) {
+	t(_, {
+		$lt: 4,
+	}, 5, false);
+
+	t(_, {
+		$lt: 5,
+	}, 5, false);
+
+	t(_, {
+		$lt: 6,
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("lte", 3, function(_) {
+	t(_, {
+		$lte: 4,
+	}, 5, false);
+
+	t(_, {
+		$lte: 5,
+	}, 5, true);
+
+	t(_, {
+		$lte: 6,
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("ne", 3, function(_) {
+	t(_, {
+		$ne: 4,
+	}, 5, true);
+
+	t(_, {
+		$ne: 5,
+	}, 5, false);
+
+	t(_, {
+		$ne: 6,
+	}, 5, true);
+
+
+	start();
+});
+
+asyncTest("range", 3, function(_) {
+	t(_, {
+		$gte: 3,
+		$lte: 7,
+	}, 2, false);
+
+	t(_, {
+		$gte: 3,
+		$lte: 7,
+	}, 5, true);
+
+	t(_, {
+		$gte: 3,
+		$lte: 7,
+	}, 8, false);
+
+	start();
+});
+
+asyncTest("regexp", 2, function(_) {
+	t(_, /^hel/, 'hello', true);
+	t(_, /^hel/, 'world', false);
+
+	start();
+});
+
+asyncTest("and", 2, function(_) {
+	t(_, {
+		$and: [2, 5],
+	}, 5, false);
+
+	t(_, {
+		$and: [5, 5],
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("or", 2, function(_) {
+	t(_, {
+		$or: [2, 5],
+	}, 5, true);
+
+	t(_, {
+		$or: [2, 6],
+	}, 5, false);
+
+	start();
+});
+
+asyncTest("nor", 2, function(_) {
+	t(_, {
+		$nor: [2, 5],
+	}, 5, false);
+
+	t(_, {
+		$nor: [2, 6],
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("not", 2, function(_) {
+	t(_, {
+		$not: {
+			$gt: 2
+		},
+	}, 5, false);
+
+	t(_, {
+		$not: {
+			$lt: 2
+		},
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("in", 3, function(_) {
+	t(_, {
+		$in: [2, 3, 5]
+	}, 3, true);
+
+	t(_, {
+		$in: [2, 3, 5]
+	}, 4, false);
+
+	t(_, {
+		$in: [2, 3, 5]
+	}, 5, true);
+
+	start();
+});
+
+asyncTest("not in", 3, function(_) {
+	t(_, {
+		$nin: [2, 3, 5]
+	}, 3, false);
+
+	t(_, {
+		$nin: [2, 3, 5]
+	}, 4, true);
+
+	t(_, {
+		$nin: [2, 3, 5]
+	}, 5, false);
+
+	start();
+});
+
+asyncTest("empty and", 2, function(_) {
+	t(_, {}, {}, true);
+
+	t(_, {}, {
+		a: 5,
+	}, true);
+
+	start();
+});
+
+asyncTest("empty or", 2, function(_) {
+	t(_, {
+		$or: []
+	}, {}, false);
+
+	t(_, {
+		$or: []
+	}, {
+		a: 5,
+	}, false);
+
+	start();
+});
+
+asyncTest("single property", 2, function(_) {
+	t(_, {
+		a: 5,
+	}, {
+		a: 5,
+		b: 3,
+	}, true);
+
+	t(_, {
+		a: 6,
+	}, {
+		a: 5,
+		b: 3,
+	}, false);
+	start();
+});
+
+asyncTest("implicit and (multiple properties)", 2, function(_) {
+	t(_, {
+		a: 5,
+		b: 3,
+	}, {
+		a: 5,
+		b: 3,
+	}, true);
+
+	t(_, {
+		a: 5,
+		b: 3,
+	}, {
+		a: 5,
+	}, false);
+
+	start();
+});
+
+asyncTest("walk", 5, function(_) {
+	t(_, {
+		'a.b': /^hel/,
+	}, {
+		a: {
+			b: 'hello',
+		}
+	}, true);
+
+	t(_, {
+		'a.b': /^hel/,
+	}, {
+		a: {
+			c: 'hello',
+		}
+	}, false);
+
+	t(_, {
+		'a.c': /^hel/,
+	}, {
+		b: {
+			c: 'hello',
+		}
+	}, false);
+
+	t(_, {
+		'a.b.c': /^hel/,
+	}, {
+		a: {
+			b: {
+				c: 'hello',
+			}
+		}
+	}, true);
+
+	t(_, {
+		'a.b.c': /^hel/,
+	}, {
+		a: {
+			b: {
+				c: 'world',
+			}
+		}
+	}, false);
+
+	start();
+});
\ No newline at end of file
diff --git a/test/server/api-test._js b/test/server/api-test._js
index 24b2cd9..edcdfb8 100644
--- a/test/server/api-test._js
+++ b/test/server/api-test._js
@@ -35,7 +35,7 @@ asyncTest("map", 1, function(_) {
 	start();
 });
 
-asyncTest("every", 3, function(_) {
+asyncTest("every", 6, function(_) {
 	strictEqual(numbers(5).every(_, function(_, num) {
 		return num < 5;
 	}), true);
@@ -45,10 +45,19 @@ asyncTest("every", 3, function(_) {
 	strictEqual(numbers(5).every(_, function(_, num) {
 		return num != 2;
 	}), false);
+	strictEqual(numbers(5).every(_, {
+		$lt: 5,
+	}), true);
+	strictEqual(numbers(5).every(_, {
+		$lt: 4,
+	}), false);
+	strictEqual(numbers(5).every(_, {
+		$ne: 2,
+	}), false);
 	start();
 });
 
-asyncTest("some", 3, function(_) {
+asyncTest("some", 6, function(_) {
 	strictEqual(numbers(5).some(_, function(_, num) {
 		return num >= 5;
 	}), false);
@@ -58,6 +67,15 @@ asyncTest("some", 3, function(_) {
 	strictEqual(numbers(5).some(_, function(_, num) {
 		return num != 2;
 	}), true);
+	strictEqual(numbers(5).some(_, {
+		$gte: 5,
+	}), false);
+	strictEqual(numbers(5).some(_, {
+		$gte: 4,
+	}), true);
+	strictEqual(numbers(5).some(_, {
+		$ne: 2,
+	}), true);
 	start();
 });
 
@@ -114,24 +132,34 @@ asyncTest("transform - less reads than writes", 1, function(_) {
 	start();
 });
 
-asyncTest("filter", 1, function(_) {
+asyncTest("filter", 2, function(_) {
 	strictEqual(numbers(10).filter(function(_, val) {
 		return val % 2;
 	}).pipe(_, arraySink()).toArray().join(','), "1,3,5,7,9");
+	strictEqual(numbers(10).filter({
+		$gt: 2,
+		$lt: 6,
+	}).pipe(_, arraySink()).toArray().join(','), "3,4,5");
 	start();
 });
 
-asyncTest("while", 1, function(_) {
+asyncTest("while", 2, function(_) {
 	strictEqual(numbers().while(function(_, val) {
 		return val < 5;
 	}).pipe(_, arraySink()).toArray().join(','), "0,1,2,3,4");
+	strictEqual(numbers().while({
+		$lt: 5,
+	}).pipe(_, arraySink()).toArray().join(','), "0,1,2,3,4");
 	start();
 });
 
-asyncTest("until", 1, function(_) {
+asyncTest("until", 2, function(_) {
 	strictEqual(numbers().until(function(_, val) {
 		return val > 5;
 	}).pipe(_, arraySink()).toArray().join(','), "0,1,2,3,4,5");
+	strictEqual(numbers().until({
+		$gt: 5,
+	}).pipe(_, arraySink()).toArray().join(','), "0,1,2,3,4,5");
 	start();
 });