feat(cache): add cache operator

Author: benlesh

spec/operators/cache-spec.js

@@ -0,0 +1,125 @@
+/* globals describe, it, expect, expectObservable, hot, rxTestScheduler */
+var Rx = require('../../dist/cjs/Rx');
+
+var asap = Rx.Scheduler.asap;
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.cache', function () {
+  it('should replay values upon subscription', function () {
+    var s1 = hot('---^---a---b---c---|     ').cache();
+    var expected1 = '----a---b---c---|     ';
+    var expected2 = '                (abc|)';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should replay values and error', function () {
+    var s1 = hot('---^---a---b---c---#     ').cache();
+    var expected1 = '----a---b---c---#     ';
+    var expected2 = '                (abc#)';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should replay values and and share', function () {
+    var s1 = hot('---^---a---b---c------------d--e--f-|').cache();
+    var expected1 = '----a---b---c------------d--e--f-|';
+    var expected2 = '                (abc)----d--e--f-|';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should have a bufferCount that limits the replay test 1', function () {
+    var s1 = hot('---^---a---b---c------------d--e--f-|').cache(1);
+    var expected1 = '----a---b---c------------d--e--f-|';
+    var expected2 = '                c--------d--e--f-|';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should have a bufferCount that limits the replay test 2', function () {
+    var s1 = hot('---^---a---b---c------------d--e--f-|').cache(2);
+    var expected1 = '----a---b---c------------d--e--f-|';
+    var expected2 = '                (bc)-----d--e--f-|';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should accept a windowTime that limits the replay', function () {
+    var w = time(         '----------|');
+    var s1 = hot('---^---a---b---c------------d--e--f-|').cache(Number.POSITIVE_INFINITY, w, rxTestScheduler);
+    var expected1 = '----a---b---c------------d--e--f-|';
+    var expected2 = '                (bc)-----d--e--f-|';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should handle empty', function () {
+    var s1 =   cold('|').cache();
+    var expected1 = '|';
+    var expected2 = '                |';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should handle throw', function () {
+    var s1 =   cold('#').cache();
+    var expected1 = '#';
+    var expected2 = '                #';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+
+  it('should handle never', function () {
+    var s1 =   cold('-').cache();
+    var expected1 = '-';
+    var expected2 = '                -';
+    var t = time(   '----------------|');
+
+    expectObservable(s1).toBe(expected1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(expected2);
+    }, t);
+  });
+});

src/Observable.ts

@@ -210,6 +210,7 @@ export class Observable<T> implements CoreOperators<T>  {
   bufferTime: BufferTimeSignature<T>;
   bufferToggle: BufferToggleSignature<T>;
   bufferWhen: BufferWhenSignature<T>;
+  cache: (bufferSize?: number, windowTime?: number, scheduler?: Scheduler) => Observable<T>;
   catch: (selector: (err: any, source: Observable<T>, caught: Observable<any>) => Observable<any>) => Observable<T>;
   combineAll: <R>(project?: (...values: Array<any>) => R) => Observable<R>;
   combineLatest: CombineLatestSignature<T>;

src/Rx.ts

@@ -35,6 +35,7 @@ import './add/operator/bufferCount';
 import './add/operator/bufferTime';
 import './add/operator/bufferToggle';
 import './add/operator/bufferWhen';
+import './add/operator/cache';
 import './add/operator/catch';
 import './add/operator/combineAll';
 import './add/operator/combineLatest';

src/add/operator/cache.ts

@@ -0,0 +1,7 @@
+
+import {Observable} from '../../Observable';
+import {cache} from '../../operator/cache';
+
+Observable.prototype.cache = cache;
+
+export var _void: void;

src/operator/cache.ts

@@ -0,0 +1,10 @@
+import {Observable} from '../Observable';
+import {publishReplay} from './publishReplay';
+import {Scheduler} from '../Scheduler';
+import {ConnectableObservable} from '../observable/ConnectableObservable';
+
+export function cache<T>(bufferSize: number = Number.POSITIVE_INFINITY,
+                         windowTime: number = Number.POSITIVE_INFINITY,
+                         scheduler?: Scheduler): Observable<T> {
+  return (<ConnectableObservable<any>>publishReplay.call(this, bufferSize, windowTime, scheduler)).refCount();
+}