Observable.count;

doc/README.md

@@ -24,6 +24,7 @@ RxLua
   - [combineLatest](#combinelatestobservables-combinator)
   - [compact](#compact)
   - [concat](#concatsources)
+  - [count](#countpredicate)
   - [distinct](#distinct)
   - [distinctUntilChanged](#distinctuntilchangedcomparator)
   - [filter](#filterpredicate)
@@ -272,6 +273,16 @@ Returns a new Observable that produces the values produced by all the specified
 
 ---
 
+#### `:count(predicate)`
+
+Returns an Observable that produces a single value representing the number of values produced by the source value that satisfy an optional predicate.
+
+| Name | Type | Default | Description |
+|------|------|---------|-------------|
+| `predicate` | function (optional) |  | The predicate used to match values. |
+
+---
+
 #### `:distinct()`
 
 Returns a new Observable that produces the values from the original with duplicates removed.

rx.lua

@@ -335,6 +335,34 @@ function Observable:concat(other, ...)
   end)
 end
 
+--- Returns an Observable that produces a single value representing the number of values produced
+-- by the source value that satisfy an optional predicate.
+-- @arg {function=} predicate - The predicate used to match values.
+function Observable:count(predicate)
+  predicate = predicate or util.constant(true)
+
+  return Observable.create(function(observer)
+    local count = 0
+
+    local function onNext(...)
+      if predicate(...) then
+        count = count + 1
+      end
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onComplete()
+      observer:onNext(count)
+      observer:onComplete()
+    end
+
+    return self:subscribe(onNext, onError, onComplete)
+  end)
+end
+
 --- Returns a new Observable that produces the values from the original with duplicates removed.
 -- @returns {Observable}
 function Observable:distinct()

src/observable.lua

@@ -250,6 +250,34 @@ function Observable:concat(other, ...)
   end)
 end
 
+--- Returns an Observable that produces a single value representing the number of values produced
+-- by the source value that satisfy an optional predicate.
+-- @arg {function=} predicate - The predicate used to match values.
+function Observable:count(predicate)
+  predicate = predicate or util.constant(true)
+
+  return Observable.create(function(observer)
+    local count = 0
+
+    local function onNext(...)
+      if predicate(...) then
+        count = count + 1
+      end
+    end
+
+    local function onError(e)
+      return observer:onError(e)
+    end
+
+    local function onComplete()
+      observer:onNext(count)
+      observer:onComplete()
+    end
+
+    return self:subscribe(onNext, onError, onComplete)
+  end)
+end
+
 --- Returns a new Observable that produces the values from the original with duplicates removed.
 -- @returns {Observable}
 function Observable:distinct()

tests/count.lua

@@ -0,0 +1,17 @@
+describe('count', function()
+  it('passes through errors', function()
+    local observable = Rx.Observable.create(function(observer) observer:onError() end)
+    expect(observable.subscribe).to.fail()
+    expect(observable:count().subscribe).to.fail()
+  end)
+
+  it('produces a single value representing the number of elements produced by the source', function()
+    local observable = Rx.Observable.fromRange(5):count()
+    expect(observable).to.produce(5)
+  end)
+
+  it('uses the predicate to filter for values if it is specified', function()
+    local observable = Rx.Observable.fromRange(5):count(function(x) return x > 3 end)
+    expect(observable).to.produce(2)
+  end)
+end)