Merge pull request #1 from maniator/scan-reduce

Create scan and reduce functionality

Author: maniator

example/index.js

@@ -35,4 +35,3 @@ inputObservable$
 inputObservable$
   .do((text) => div.textContent = text)
   .subscribe();
-

src/index.js

@@ -15,6 +15,8 @@ export * from './operators/first';
 export * from './operators/toPromise';
 export * from './operators/flatMap';
 export * from './operators/switchMap';
+export * from './operators/scan';
+export * from './operators/reduce';
 
 // add on observable types
 export * from './observables/fromEvent';

src/operators/reduce.js

@@ -0,0 +1,24 @@
+import { Observable } from '../Observable';
+import { scan } from './scan';
+
+export const reduce = function (source$, scanCallback, startValue = 0) {
+  return new Observable(function ({next, error, complete }) {
+    let lastValue = 0;
+    const subscription = scan(source$, scanCallback, startValue).subscribe({
+      next (value) {
+        lastValue = value;
+      },
+      error,
+      complete () {
+        next(lastValue);
+        complete();
+      }
+    });
+    
+    return () => subscription.unsubscribe();
+  });
+};
+
+Observable.prototype.reduce = function (scanCallback, startValue) {
+  return reduce(this, scanCallback, startValue);
+};

src/operators/scan.js

@@ -0,0 +1,23 @@
+import { Observable } from '../Observable';
+
+export const scan = function (source$, scanCallback, startValue = 0) {
+  return new Observable(function ({ next, error, complete }) {
+    let previousValue = startValue;
+    
+    const subscription = source$.subscribe({
+      next (value) {
+        previousValue = scanCallback(previousValue, value);
+        
+        next(previousValue);
+      },
+      error,
+      complete
+    });
+    
+    return () => subscription.unsubscribe();
+  });
+};
+
+Observable.prototype.scan = function (scanCallback, startValue) {
+  return scan(this, scanCallback, startValue);
+};