Add Observable.add;

doc/README.md

@@ -17,6 +17,7 @@ RxLua
   - [fromTable](#fromtabletable-iterator-keys)
   - [fromCoroutine](#fromcoroutinecoroutine)
   - [dump](#dumpname-formatter)
+  - [all](#allpredicate)
   - [combineLatest](#combinelatestobservables-combinator)
   - [compact](#compact)
   - [concat](#concatsources)
@@ -213,6 +214,16 @@ Subscribes to this Observable and prints values it produces.
 
 ---
 
+#### `:all(predicate)`
+
+Determine whether all items emitted by an Observable meet some criteria.
+
+| Name | Type | Default | Description |
+|------|------|---------|-------------|
+| `predicate` | function (optional) | identity | The predicate used to evaluate objects. |
+
+---
+
 #### `:combineLatest(observables, combinator)`
 
 Returns a new Observable that runs a combinator function on the most recent values from a set of Observables whenever any of them produce a new value. The results of the combinator function are produced by the new Observable.

src/observable.lua

@@ -117,6 +117,32 @@ function Observable:dump(name, formatter)
   return self:subscribe(onNext, onError, onComplete)
 end
 
+--- Determine whether all items emitted by an Observable meet some criteria.
+-- @arg {function=identity} predicate - The predicate used to evaluate objects.
+function Observable:all(predicate)
+  predicate = predicate or util.identity
+
+  return Observable.create(function(observer)
+    local function onNext(...)
+      if not predicate(...) then
+        observer:onNext(false)
+        observer:onComplete()
+      end
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onComplete()
+      observer:onNext(true)
+      return observer:onComplete()
+    end
+
+    return self:subscribe(onNext, onError, onComplete)
+  end)
+end
+
 --- Returns a new Observable that runs a combinator function on the most recent values from a set
 -- of Observables whenever any of them produce a new value. The results of the combinator function
 -- are produced by the new Observable.

tests/all.lua

@@ -0,0 +1,25 @@
+describe('all', function()
+  it('passes through errors', function()
+    local observable = Rx.Observable.create(function(observer) observer:onError() end)
+    expect(observable.subscribe).to.fail()
+    expect(observable:all().subscribe).to.fail()
+  end)
+
+  it('produces true if all elements satisfy the predicate', function()
+    local observable = Rx.Observable.fromRange(5):all(function(x) return x < 10 end)
+    expect(observable).to.produce({{true}})
+  end)
+
+  it('produces false if one element does not satisfy the predicate', function()
+    local observable = Rx.Observable.fromRange(5):all(function(x) return x ~= 3 end)
+    expect(observable).to.produce({{false}})
+  end)
+
+  it('uses the identity function as a predicate if none is specified', function()
+    local observable = Rx.Observable.fromValue(false):all()
+    expect(observable).to.produce({{false}})
+
+    observable = Rx.Observable.fromValue(true):all()
+    expect(observable).to.produce({{true}})
+  end)
+end)

tests/observable.lua

@@ -150,6 +150,7 @@ describe('Observable', function()
   describe('dump', function()
   end)
 
+  dofile('tests/all.lua')
   dofile('tests/combineLatest.lua')
   dofile('tests/compact.lua')
   dofile('tests/concat.lua')