feat(share): add the share operator

closes #439

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

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

perf/micro/index.js

@@ -29,7 +29,7 @@ Observable.create(function(observer) {
 })
 .filter(function(filePath) {
   var argv = process.argv;
-  if(argv && argv.length > 2) {    
+  if(argv && argv.length > 2) {   
     return argv.slice(2).some(function(val) {
       return path.parse(filePath).name === val;
     });

spec/operators/share-spec.js

@@ -0,0 +1,64 @@
+/* globals describe, expect, it, hot, cold, expectObservable */
+
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.share()', function (){
+  it('should share a single subscription', function (){
+    var subscriptionCount = 0;
+    var obs = new Observable(function(observer){
+      subscriptionCount++;
+    });
+
+    var source = obs.share();
+
+    expect(subscriptionCount).toBe(0);
+
+    source.subscribe();
+    source.subscribe();
+
+    expect(subscriptionCount).toBe(1);
+  });
+
+  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.share()).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.share()).toBe(expected);
+  });
+
+   it('should not change the output of the observable when successful with cold observable', function (){ 
+    var e1 =  cold('---a--b--c--d--e--|');
+    var expected = '---a--b--c--d--e--|';
+
+    expectObservable(e1.share()).toBe(expected);
+  });
+
+  it('should not change the output of the observable when error with cold observable', function (){ 
+    var e1 =  cold('---a--b--c--d--e--#');
+    var expected = '---a--b--c--d--e--#';
+
+    expectObservable(e1.share()).toBe(expected);
+  });
+
+  it('should not change the output of the observable when never', function (){ 
+    var e1 = Observable.never();
+    var expected = '-';
+
+    expectObservable(e1.share()).toBe(expected);
+  });
+
+  it('should not change the output of the observable when empty', function (){ 
+    var e1 = Observable.empty();
+    var expected = '|';
+
+    expectObservable(e1.share()).toBe(expected);
+  });
+});

src/CoreOperators.ts

@@ -54,6 +54,7 @@ export interface CoreOperators<T> {
   sample?: <T>(notifier: Observable<any>) => Observable<T>;
   sampleTime?: <T>(delay: number, scheduler?: Scheduler) => Observable<T>;
   scan?: <R>(project: (acc: R, x: T) => R, acc?: R) => Observable<R>;
+  share?: () => 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

@@ -223,6 +223,9 @@ observableProto.sampleTime = sampleTime;
 import scan from './operators/scan';
 observableProto.scan = scan;
 
+import share from './operators/share';
+observableProto.share = share;
+
 import single from './operators/single';
 observableProto.single = single;
 

src/Rx.ts

@@ -203,6 +203,9 @@ observableProto.sampleTime = sampleTime;
 import scan from './operators/scan';
 observableProto.scan = scan;
 
+import share from './operators/share';
+observableProto.share = share;
+
 import single from './operators/single';
 observableProto.single = single;
 

src/operators/share.ts

@@ -0,0 +1,6 @@
+import Observable from '../Observable';
+import publish from './publish';
+
+export default function share<T>() : Observable<T> {
+  return publish.call(this).refCount();
+};