fix(publish): fix selector typings (#2891)

Fixes the typings for publish and multicast, so that selectors that
change the observable's type are supported.

Closes #2889

spec/operators/multicast-spec.ts

@@ -3,6 +3,7 @@ import * as Rx from '../../dist/package/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
 declare const { asDiagram, time, rxTestScheduler };
+declare const type;
 declare const hot: typeof marbleTestingSignature.hot;
 declare const cold: typeof marbleTestingSignature.cold;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
@@ -638,4 +639,27 @@ describe('Observable.prototype.multicast', () => {
         });
     });
   });
+
+  describe('typings', () => {
+    type('should infer the type', () => {
+      /* tslint:disable:no-unused-variable */
+      const source = Rx.Observable.of<number>(1, 2, 3);
+      const result: Rx.Observable<number> = source.multicast(() => new Subject<number>());
+      /* tslint:enable:no-unused-variable */
+    });
+
+    type('should infer the type with a selector', () => {
+      /* tslint:disable:no-unused-variable */
+      const source = Rx.Observable.of<number>(1, 2, 3);
+      const result: Rx.Observable<number> = source.multicast(() => new Subject<number>(), s => s.map(x => x));
+      /* tslint:enable:no-unused-variable */
+    });
+
+    type('should infer the type with a type-changing selector', () => {
+      /* tslint:disable:no-unused-variable */
+      const source = Rx.Observable.of<number>(1, 2, 3);
+      const result: Rx.Observable<string> = source.multicast(() => new Subject<number>(), s => s.map(x => x + '!'));
+      /* tslint:enable:no-unused-variable */
+    });
+  });
 });

spec/operators/publish-spec.ts

@@ -3,6 +3,7 @@ import * as Rx from '../../dist/package/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
 declare const { asDiagram };
+declare const type;
 declare const hot: typeof marbleTestingSignature.hot;
 declare const cold: typeof marbleTestingSignature.cold;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
@@ -332,4 +333,25 @@ describe('Observable.prototype.publish', () => {
     expect(subscriptions).to.equal(1);
     done();
   });
+
+  type('should infer the type', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.Observable<number> = source.publish();
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type with a selector', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.Observable<number> = source.publish(s => s.map(x => x));
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type with a type-changing selector', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.Observable<string> = source.publish(s => s.map(x => x + '!'));
+    /* tslint:enable:no-unused-variable */
+  });
 });

src/operator/multicast.ts

@@ -2,11 +2,12 @@ import { Subject } from '../Subject';
 import { Observable } from '../Observable';
 import { ConnectableObservable } from '../observable/ConnectableObservable';
 import { multicast as higherOrder } from '../operators/multicast';
-import { FactoryOrValue, MonoTypeOperatorFunction } from '../interfaces';
+import { FactoryOrValue, MonoTypeOperatorFunction, OperatorFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
 export function multicast<T>(this: Observable<T>, subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): ConnectableObservable<T>;
 export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: MonoTypeOperatorFunction<T>): Observable<T>;
+export function multicast<T, R>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: OperatorFunction<T, R>): Observable<R>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -103,7 +104,7 @@ export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>
  * @method multicast
  * @owner Observable
  */
-export function multicast<T>(this: Observable<T>, subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
-                             selector?: (source: Observable<T>) => Observable<T>): Observable<T> | ConnectableObservable<T> {
+export function multicast<T, R>(this: Observable<T>, subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
+                                selector?: (source: Observable<T>) => Observable<R>): Observable<R> | ConnectableObservable<R> {
   return higherOrder(<any>subjectOrSubjectFactory, selector)(this);
 }

src/operator/publish.ts

@@ -5,7 +5,8 @@ import { publish as higherOrder } from '../operators/publish';
 
 /* tslint:disable:max-line-length */
 export function publish<T>(this: Observable<T>): ConnectableObservable<T>;
-export function publish<T>(this: Observable<T>, selector: selector<T>): Observable<T>;
+export function publish<T>(this: Observable<T>, selector: (source: Observable<T>) => Observable<T>): Observable<T>;
+export function publish<T, R>(this: Observable<T>, selector: (source: Observable<T>) => Observable<R>): Observable<R>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -21,7 +22,7 @@ export function publish<T>(this: Observable<T>, selector: selector<T>): Observab
  * @method publish
  * @owner Observable
  */
-export function publish<T>(this: Observable<T>, selector?: (source: Observable<T>) => Observable<T>): Observable<T> | ConnectableObservable<T> {
+export function publish<T, R>(this: Observable<T>, selector?: (source: Observable<T>) => Observable<R>): Observable<R> | ConnectableObservable<R> {
   return higherOrder(selector)(this);
 }
 

src/operators/multicast.ts

@@ -3,11 +3,12 @@ import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
 import { ConnectableObservable, connectableObservableDescriptor } from '../observable/ConnectableObservable';
-import { FactoryOrValue, MonoTypeOperatorFunction } from '../interfaces';
+import { FactoryOrValue, MonoTypeOperatorFunction, OperatorFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
 export function multicast<T>(subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): MonoTypeOperatorFunction<T>;
 export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
+export function multicast<T, R>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: OperatorFunction<T, R>): OperatorFunction<T, R>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -29,9 +30,9 @@ export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>
  * @method multicast
  * @owner Observable
  */
-export function multicast<T>(subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
-                             selector?: (source: Observable<T>) => Observable<T>): MonoTypeOperatorFunction<T> {
-  return function multicastOperatorFunction(source: Observable<T>): Observable<T> {
+export function multicast<T, R>(subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
+                                selector?: (source: Observable<T>) => Observable<R>): OperatorFunction<T, R> {
+  return function multicastOperatorFunction(source: Observable<T>): Observable<R> {
     let subjectFactory: () => Subject<T>;
     if (typeof subjectOrSubjectFactory === 'function') {
       subjectFactory = <() => Subject<T>>subjectOrSubjectFactory;
@@ -49,15 +50,15 @@ export function multicast<T>(subjectOrSubjectFactory: Subject<T> | (() => Subjec
     connectable.source = source;
     connectable.subjectFactory = subjectFactory;
 
-    return <ConnectableObservable<T>> connectable;
+    return <ConnectableObservable<R>> connectable;
   };
 }
 
-export class MulticastOperator<T> implements Operator<T, T> {
+export class MulticastOperator<T, R> implements Operator<T, R> {
   constructor(private subjectFactory: () => Subject<T>,
-              private selector: (source: Observable<T>) => Observable<T>) {
+              private selector: (source: Observable<T>) => Observable<R>) {
   }
-  call(subscriber: Subscriber<T>, source: any): any {
+  call(subscriber: Subscriber<R>, source: any): any {
     const { selector } = this;
     const subject = this.subjectFactory();
     const subscription = selector(subject).subscribe(subscriber);

src/operators/publish.ts

@@ -1,10 +1,11 @@
 import { Subject } from '../Subject';
 import { multicast } from './multicast';
-import { MonoTypeOperatorFunction } from '../interfaces';
+import { MonoTypeOperatorFunction, OperatorFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
 export function publish<T>(): MonoTypeOperatorFunction<T>;
 export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
+export function publish<T, R>(selector: OperatorFunction<T, R>): OperatorFunction<T, R>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -20,7 +21,7 @@ export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOpera
  * @method publish
  * @owner Observable
  */
-export function publish<T>(selector?: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T> {
+export function publish<T, R>(selector?: OperatorFunction<T, R>): MonoTypeOperatorFunction<T> | OperatorFunction<T, R> {
   return selector ?
     multicast(() => new Subject<T>(), selector) :
     multicast(new Subject<T>());