feat(publish): support optional selectors

closes ReactiveX#1629

spec/operators/publish-spec.ts

@@ -77,6 +77,23 @@ describe('Observable.prototype.publish', () => {
  published.connect();
  });
 
+ it('should accept selectors', () => {
+ const source = hot('-1-2-3----4-|');
+ const sourceSubs = ['^ !'];
+ const published = source.publish(x => x.zip(x, (a, b) => (parseInt(a) + parseInt(b)).toString()));
+ const subscriber1 = hot('a| ').mergeMapTo(published);
+ const expected1 = '-2-4-6----8-|';
+ const subscriber2 = hot(' b| ').mergeMapTo(published);
+ const expected2 = ' -6----8-|';
+ const subscriber3 = hot(' c| ').mergeMapTo(published);
+ const expected3 = ' --8-|';
+
+ expectObservable(subscriber1).toBe(expected1);
+ expectObservable(subscriber2).toBe(expected2);
+ expectObservable(subscriber3).toBe(expected3);
+ expectSubscriptions(source.subscriptions).toBe(sourceSubs);
+ });
+
  it('should multicast an error from the source to multiple observers', () => {
  const source = cold('-1-2-3----4-#');
  const sourceSubs = '^ !';

src/add/operator/publish.ts

@@ -2,7 +2,7 @@
 import {Observable} from '../../Observable';
 import {publish, PublishSignature} from '../../operator/publish';
 
-Observable.prototype.publish = publish;
+Observable.prototype.publish = <any>publish;
 
 declare module '../../Observable' {
  interface Observable<T> {

src/operator/publish.ts

@@ -1,4 +1,5 @@
 import {Subject} from '../Subject';
+import {Observable} from '../Observable';
 import {multicast} from './multicast';
 import {ConnectableObservable} from '../observable/ConnectableObservable';
 
@@ -8,14 +9,21 @@ import {ConnectableObservable} from '../observable/ConnectableObservable';
  *
  * <img src="./img/publish.png" width="100%">
  *
+ * @param {Function} Optional selector function which can use the multicasted source sequence as many times as needed,
+ * without causing multiple subscriptions to the source sequence.
+ * Subscribers to the given source will receive all notifications of the source from the time of the subscription on.
  * @return a ConnectableObservable that upon connection causes the source Observable to emit items to its Observers.
  * @method publish
  * @owner Observable
  */
-export function publish<T>(): ConnectableObservable<T> {
- return multicast.call(this, new Subject<T>());
+export function publish<T>(selector?: (source: Observable<T>) => Observable<T>): Observable<T> | ConnectableObservable<T> {
+ return selector ? multicast.call(this, () => new Subject<T>(), selector) :
+ multicast.call(this, new Subject<T>());
 }
 
+export type selector<T> = (source: Observable<T>) => Observable<T>;
+
 export interface PublishSignature<T> {
  (): ConnectableObservable<T>;
+ (selector: selector<T>): Observable<T>;
 }