feat(shareReplay): add the shareReplay() operator

Add shareReplay() as a core operator. Vaguely relates to issue #439 and
issue #298.

Author: Andre Medeiros

perf/micro/immediate-scheduler/operators/share-replay.js

@@ -0,0 +1,20 @@
+var RxOld = require('rx');
+var RxNew = require('../../../../index');
+
+module.exports = function (suite) {
+  var oldShareReplayWithImmediateScheduler = RxOld.Observable.range(0, 25, RxOld.Scheduler.immediate)
+    .shareReplay(3);
+  var newShareReplayWithImmediateScheduler = RxNew.Observable.range(0, 25)
+    .shareReplay(3);
+
+  function _next(x) { }
+  function _error(e) { }
+  function _complete() { }
+  return suite
+    .add('old shareReplay with immediate scheduler', function () {
+      oldShareReplayWithImmediateScheduler.subscribe(_next, _error, _complete);
+    })
+    .add('new shareReplay with immediate scheduler', function () {
+      newShareReplayWithImmediateScheduler.subscribe(_next, _error, _complete);
+    });
+};

spec/operators/shareReplay-spec.js

@@ -0,0 +1,85 @@
+/* globals describe, expect, it, hot, cold, expectObservable */
+
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.shareReplay()', function () {
+  it('should share a single subscription', function () {
+    var subscriptionCount = 0;
+    var obs = new Observable(function (observer) {
+      subscriptionCount++;
+    });
+
+    var source = obs.shareReplay(1);
+
+    expect(subscriptionCount).toBe(0);
+
+    source.subscribe();
+    source.subscribe();
+
+    expect(subscriptionCount).toBe(1);
+  });
+
+  it('should replay as many events as specified by the bufferSize', function (done) {
+    var results1 = [];
+    var results2 = [];
+    var subscriptions = 0;
+
+    var source = new Observable(function (observer) {
+      subscriptions++;
+      observer.next(1);
+      observer.next(2);
+      observer.next(3);
+      observer.next(4);
+    });
+
+    var hot = source.shareReplay(2);
+
+    expect(results1).toEqual([]);
+    expect(results2).toEqual([]);
+
+    hot.subscribe(function (x) {
+      results1.push(x);
+    });
+
+    expect(results1).toEqual([1, 2, 3, 4]);
+    expect(results2).toEqual([]);
+
+    hot.subscribe(function (x) {
+      results2.push(x);
+    });
+
+    expect(results1).toEqual([1, 2, 3, 4]);
+    expect(results2).toEqual([3, 4]);
+    expect(subscriptions).toBe(1);
+    done();
+  });
+
+  it('should not change the output of the observable when successful', function () {
+    var e1 = hot('---a--^--b-c--d--e--|');
+    var expected =     '---b-c--d--e--|';
+
+    expectObservable(e1.shareReplay(1)).toBe(expected);
+  });
+
+  it('should not change the output of the observable when error', function () {
+    var e1 = hot('---a--^--b-c--d--e--#');
+    var expected =     '---b-c--d--e--#';
+
+    expectObservable(e1.shareReplay(1)).toBe(expected);
+  });
+
+  it('should not change the output of the observable when never', function () {
+    var e1 = Observable.never();
+    var expected = '-';
+
+    expectObservable(e1.shareReplay(1)).toBe(expected);
+  });
+
+  it('should not change the output of the observable when empty', function () {
+    var e1 = Observable.empty();
+    var expected = '|';
+
+    expectObservable(e1.shareReplay(1)).toBe(expected);
+  });
+});

src/CoreOperators.ts

@@ -54,6 +54,7 @@ export interface CoreOperators<T> {
   sampleTime?: <T>(delay: number, scheduler?: Scheduler) => Observable<T>;
   scan?: <R>(project: (acc: R, x: T) => R, acc?: R) => Observable<R>;
   share?: () => Observable<T>;
+  shareReplay?: (bufferSize: number, windowTime: number, scheduler?: Scheduler) => Observable<T>;
   single?: (predicate?: (value: T, index:number) => boolean, thisArg?: any) => Observable<T>;
   skip?: (count: number) => Observable<T>;
   skipUntil?: (notifier: Observable<any>) => Observable<T>;

src/Rx.KitchenSink.ts

@@ -66,7 +66,7 @@ Observable.zip = zipStatic
 
 
 
-// Operators 
+// Operators
 const observableProto = (<KitchenSinkOperators<any>>Observable.prototype);
 
 import buffer from './operators/buffer';
@@ -227,6 +227,9 @@ observableProto.scan = scan;
 import share from './operators/share';
 observableProto.share = share;
 
+import shareReplay from './operators/shareReplay';
+observableProto.shareReplay = shareReplay;
+
 import single from './operators/single';
 observableProto.single = single;
 

src/Rx.ts

@@ -56,7 +56,7 @@ Observable.zip = zipStatic
 
 
 
-// Operators 
+// Operators
 import { CoreOperators } from './CoreOperators';
 const observableProto = (<CoreOperators<any>>Observable.prototype);
 
@@ -203,6 +203,9 @@ observableProto.scan = scan;
 import share from './operators/share';
 observableProto.share = share;
 
+import shareReplay from './operators/shareReplay';
+observableProto.shareReplay = shareReplay;
+
 import single from './operators/single';
 observableProto.single = single;
 

src/operators/shareReplay.ts

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