feat(): add reduce() to extended signal

Add a reduce() method to the ReadableSignal interface, which behaves
similarly to Array.prototype.reduce().

README.md

@@ -372,6 +372,28 @@ mappedSignal.add(payload => {
 assert.deepEqual(received, ['cat!', 'dog!', 'frog!', 'sloth!']);
 ```
 
+#### Signal.reduce
+
+Signal.reduce provides the ability to aggregate payloads coming through a Signal, similar to reducing an array in JavaScript.
+
+```ts
+import {Signal} from 'micro-signals';
+import * as assert from 'assert';
+
+const signal = new Signal<number>();
+const sumSignal = signal.reduce((total, current) => total + current, 0);
+
+const received: number[] = [];
+
+sumSignal.add(payload => {
+    received.push(payload);
+});
+
+[5, 10, 20, 100].forEach(x => signal.dispatch(x));
+
+assert.deepEqual(received, [5, 15, 35, 135]);
+```
+
 #### Signal.merge
 
 Signal.merge takes an arbitrary number of signals as constructor arguments and forward payloads from

src/extended-signal.ts

@@ -1,4 +1,4 @@
-import {BaseSignal, Cache, Listener, ReadableSignal} from './interfaces';
+import {Accumulator, BaseSignal, Cache, Listener, ReadableSignal} from './interfaces';
 import { TagMap } from './tag-map';
 
 export class ExtendedSignal<T> implements ReadableSignal<T> {
@@ -103,6 +103,18 @@ export class ExtendedSignal<T> implements ReadableSignal<T> {
             listener => payload => listener(payload),
         );
     }
+    public reduce<U>(accumulator: Accumulator<T, U>, initialValue: U): ReadableSignal<U> {
+        return convertedListenerSignal(
+            this._baseSignal,
+            listener => (() => {
+                let accum = initialValue;
+                return (payload: T) => {
+                    accum = accumulator(accum, payload);
+                    listener(accum);
+                };
+            })(),
+        );
+    }
     public cache(cache: Cache<T>): ReadableSignal<T> {
         this._baseSignal.add(payload => cache.add(payload));
 

src/interfaces.ts

@@ -14,6 +14,7 @@ export interface ReadableSignal<T> extends BaseSignal<T> {
     promisify(rejectSignal?: ReadableSignal<any>): Promise<T>;
     readOnly(): ReadableSignal<T>;
     cache(cache: Cache<T>): ReadableSignal<T>;
+    reduce<U>(accumulator: Accumulator<T, U>, initialValue: U): ReadableSignal<U>;
 }
 
 export interface WritableSignal<T> {
@@ -28,3 +29,5 @@ export interface Cache<T> {
     add(payload: T): void;
     forEach(callback: (payload: T) => void): void;
 }
+
+export type Accumulator<T, U> = (accum: U, current: T) => any;

test/signal.spec.ts

@@ -10,6 +10,7 @@ import {mappedSuite} from './suites/mapped-suite';
 import {mergedSuite} from './suites/merged-suite';
 import {promisifySuite} from './suites/promisify-suite';
 import {readOnlySuite} from './suites/read-only-suite';
+import {reducedSuite} from './suites/reduced-suite';
 
 // TODO run the signal suite on the converted signals as well?
 
@@ -48,6 +49,11 @@ readOnlySuite(
     signal => signal.readOnly(),
 );
 
+reducedSuite(
+    'Signal#reduce',
+    (baseSignal, accumulator, initialValue) => baseSignal.reduce(accumulator, initialValue),
+);
+
 test('Signal listeners should received dispatched payloads', t => {
     const signal = new Signal<string>();
 

test/suites/reduced-suite.ts

@@ -0,0 +1,52 @@
+import test = require('tape');
+import {Accumulator, ReadableSignal, Signal} from '../../src';
+import {LeakDetectionSignal} from '../lib/leak-detection-signal';
+import {parentChildSuite} from './parent-child-suite';
+
+export type ReducedSignalCreationFunction = <T, U>(
+    baseSignal: ReadableSignal<T>,
+    accumulator: Accumulator<T, U>,
+    initialValue: U,
+) => ReadableSignal<U>;
+
+export function reducedSuite(prefix: string, createReducedSignal: ReducedSignalCreationFunction) {
+    parentChildSuite(prefix, () => {
+        const parentSignal = new Signal();
+        const childSignal = createReducedSignal(parentSignal, (_, payload) => payload, undefined);
+        return { parentSignal, childSignal };
+    });
+
+    test(`${prefix} should dispatch with the accumulated payload`, t => {
+        const baseSignal = new Signal<number>();
+
+        const reducedSignal = baseSignal.reduce((accum, curr) => accum + curr, 5);
+
+        const addResults: number[] = [];
+        const addOnceResults: number[] = [];
+
+        reducedSignal.add(x => addResults.push(x));
+        reducedSignal.addOnce(x => addOnceResults.push(x));
+
+        baseSignal.dispatch(50);
+        baseSignal.dispatch(0);
+        baseSignal.dispatch(100);
+
+        t.deepEqual(addResults, [55, 55, 155]);
+        t.deepEqual(addOnceResults, [55]);
+
+        t.end();
+    });
+
+    test('ReducedSignal should not leak', t => {
+        const signal = new LeakDetectionSignal<void>();
+        const mappedSignal = createReducedSignal(signal, () => true, false);
+
+        const listener = () => { /* empty listener */ };
+        mappedSignal.add(listener);
+        signal.dispatch(undefined);
+        mappedSignal.remove(listener);
+
+        t.equal(signal.listenerCount, 0);
+        t.end();
+    });
+}