feat(cache): add cache operator

spec/operators/cache-spec.js

@@ -0,0 +1,182 @@
+/* globals describe, it, expect, expectObservable, hot, rxTestScheduler, time */
+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);
+  });
+
+  it('should multicast a completion', function () {
+    var s1 = hot('--a--^--b------c-----d------e-|').cache();
+    var t1 = time(    '|                         ');
+    var e1 =          '---b------c-----d------e-|';
+    var t2 = time(    '----------|               ');
+    var e2 =          '          (bc)--d------e-|';
+    var t3 = time(    '----------------|         ');
+    var e3 =          '                (bcd)--e-|';
+
+    var expected = [e1, e2, e3];
+
+    [t1, t2, t3].forEach(function (t, i) {
+      rxTestScheduler.schedule(function () {
+        expectObservable(s1).toBe(expected[i]);
+      }, t);
+    });
+  });
+
+  it('should multicast an error', function () {
+    var s1 = hot('--a--^--b------c-----d------e-#').cache();
+    var t1 = time(    '|                         ');
+    var e1 =          '---b------c-----d------e-#';
+    var t2 = time(    '----------|               ');
+    var e2 =          '          (bc)--d------e-#';
+    var t3 = time(    '----------------|         ');
+    var e3 =          '                (bcd)--e-#';
+
+    var expected = [e1, e2, e3];
+
+    [t1, t2, t3].forEach(function (t, i) {
+      rxTestScheduler.schedule(function () {
+        expectObservable(s1).toBe(expected[i]);
+      }, t);
+    });
+  });
+
+  it('should limit replay by both count and a window time, test 2', function () {
+    var w = time(     '-----------|');
+    var s1 = hot('--a--^---b---c----d----e------f--g--h--i-------|').cache(4, w, rxTestScheduler);
+    var e1 =          '----b---c----d----e------f--g--h--i-------|';
+    var t1 = time(    '--------------------|');
+    //                          -----------|
+    var e2 =          '                    (de)-f--g--h--i-------|'; // time wins
+    var t2 = time(    '-----------------------------------|');
+    var e3 =          '                                   (fghi)-|'; // count wins
+    //                                         -----------|
+
+    expectObservable(s1).toBe(e1);
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(e2);
+    }, t1);
+
+    rxTestScheduler.schedule(function () {
+      expectObservable(s1).toBe(e3);
+    }, t2);
+  });
+});

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();
+}

src/subject/ReplaySubject.ts

@@ -9,15 +9,15 @@ export class ReplaySubject<T> extends Subject<T> {
   private events: ReplayEvent<T>[] = [];
   private scheduler: Scheduler;
   private bufferSize: number;
-  private windowSize: number;
+  private _windowTime: number;
 
   constructor(bufferSize: number = Number.POSITIVE_INFINITY,
-              windowSize: number = Number.POSITIVE_INFINITY,
+              windowTime: number = Number.POSITIVE_INFINITY,
               scheduler?: Scheduler) {
     super();
     this.scheduler = scheduler;
     this.bufferSize = bufferSize < 1 ? 1 : bufferSize;
-    this.windowSize = windowSize < 1 ? 1 : windowSize;
+    this._windowTime = windowTime < 1 ? 1 : windowTime;
   }
 
   protected _next(value: T): void {
@@ -49,7 +49,7 @@ export class ReplaySubject<T> extends Subject<T> {
 
   private _trimBufferThenGetEvents(now: number): ReplayEvent<T>[] {
     const bufferSize = this.bufferSize;
-    const windowSize = this.windowSize;
+    const _windowTime = this._windowTime;
     const events = this.events;
 
     let eventsCount = events.length;
@@ -59,7 +59,7 @@ export class ReplaySubject<T> extends Subject<T> {
     // Start at the front of the list. Break early once
     // we encounter an event that falls within the window.
     while (spliceCount < eventsCount) {
-      if ((now - events[spliceCount].time) < windowSize) {
+      if ((now - events[spliceCount].time) < _windowTime) {
         break;
       }
       spliceCount += 1;