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