Observable.average;

doc/README.md

@@ -22,6 +22,7 @@ RxLua
   - [dump](#dumpname-formatter)
   - [all](#allpredicate)
   - [amb](#ambobservables)
+  - [average](#average)
   - [combineLatest](#combinelatestobservables-combinator)
   - [compact](#compact)
   - [concat](#concatsources)
@@ -257,6 +258,12 @@ Given a set of Observables, produces values from only the first one to produce a
 
 ---
 
+#### `:average()`
+
+Returns an Observable that produces the average of all values produced by the original.
+
+---
+
 #### `: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.

rx.lua

@@ -296,6 +296,33 @@ function Observable.amb(a, b, ...)
   end):amb(...)
 end
 
+--- Returns an Observable that produces the average of all values produced by the original.
+-- @returns {Observable}
+function Observable:average()
+  return Observable.create(function(observer)
+    local sum, count = 0, 0
+
+    local function onNext(value)
+      sum = sum + value
+      count = count + 1
+    end
+
+    local function onError(e)
+      observer:onError(e)
+    end
+
+    local function onCompleted()
+      if count > 0 then
+        observer:onNext(sum / count)
+      end
+
+      observer:onCompleted()
+    end
+
+    return self:subscribe(onNext, onError, onCompleted)
+  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.

src/observable.lua

@@ -211,6 +211,33 @@ function Observable.amb(a, b, ...)
   end):amb(...)
 end
 
+--- Returns an Observable that produces the average of all values produced by the original.
+-- @returns {Observable}
+function Observable:average()
+  return Observable.create(function(observer)
+    local sum, count = 0, 0
+
+    local function onNext(value)
+      sum = sum + value
+      count = count + 1
+    end
+
+    local function onError(e)
+      observer:onError(e)
+    end
+
+    local function onCompleted()
+      if count > 0 then
+        observer:onNext(sum / count)
+      end
+
+      observer:onCompleted()
+    end
+
+    return self:subscribe(onNext, onError, onCompleted)
+  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/average.lua

@@ -0,0 +1,14 @@
+describe('average', function()
+  it('errors when its parent errors', function()
+    expect(Rx.Observable.throw():average().subscribe).to.fail()
+  end)
+
+  it('produces a single value representing the average of the values produced by the source', function()
+    expect(Rx.Observable.fromRange(3, 9, 2):average()).to.produce(6)
+    expect(Rx.Observable.fromTable({-1, 0, 1}):average()).to.produce(0)
+  end)
+
+  it('produces nothing if there are no values to average', function()
+    expect(Rx.Observable.empty():average()).to.produce({})
+  end)
+end)