fix(typings): publish variants will properly return ConnectableObserv…

…able(#2983)

* chore(typings): improve publish typings

Use ConnectableObservable in signatures that do not have selectors.

* chore(typings): add publish type tests

spec/operators/multicast-spec.ts

@@ -644,7 +644,7 @@ describe('Observable.prototype.multicast', () => {
     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>());
+      const result: Rx.ConnectableObservable<number> = source.multicast(() => new Subject<number>());
       /* tslint:enable:no-unused-variable */
     });
 
@@ -661,5 +661,27 @@ describe('Observable.prototype.multicast', () => {
       const result: Rx.Observable<string> = source.multicast(() => new Subject<number>(), s => s.map(x => x + '!'));
       /* tslint:enable:no-unused-variable */
     });
+
+    type('should infer the type for the pipeable operator', () => {
+      /* tslint:disable:no-unused-variable */
+      const source = Rx.Observable.of<number>(1, 2, 3);
+      // TODO: https://github.com/ReactiveX/rxjs/issues/2972
+      const result: Rx.ConnectableObservable<number> = Rx.operators.multicast(() => new Subject<number>())(source);
+      /* tslint:enable:no-unused-variable */
+    });
+
+    type('should infer the type for the pipeable operator with a selector', () => {
+      /* tslint:disable:no-unused-variable */
+      const source = Rx.Observable.of<number>(1, 2, 3);
+      const result: Rx.Observable<number> = source.pipe(Rx.operators.multicast(() => new Subject<number>(), s => s.map(x => x)));
+      /* tslint:enable:no-unused-variable */
+    });
+
+    type('should infer the type for the pipeable operator 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.pipe(Rx.operators.multicast(() => new Subject<number>(), s => s.map(x => x + '!')));
+      /* tslint:enable:no-unused-variable */
+    });
   });
 });

spec/operators/publish-spec.ts

@@ -337,7 +337,7 @@ describe('Observable.prototype.publish', () => {
   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();
+    const result: Rx.ConnectableObservable<number> = source.publish();
     /* tslint:enable:no-unused-variable */
   });
 
@@ -354,4 +354,26 @@ describe('Observable.prototype.publish', () => {
     const result: Rx.Observable<string> = source.publish(s => s.map(x => x + '!'));
     /* tslint:enable:no-unused-variable */
   });
+
+  type('should infer the type for the pipeable operator', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    // TODO: https://github.com/ReactiveX/rxjs/issues/2972
+    const result: Rx.ConnectableObservable<number> = Rx.operators.publish()(source);
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator with a selector', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.Observable<number> = source.pipe(Rx.operators.publish(s => s.map(x => x)));
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator 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.pipe(Rx.operators.publish(s => s.map(x => x + '!')));
+    /* tslint:enable:no-unused-variable */
+  });
 });

spec/operators/publishBehavior-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;
@@ -336,4 +337,19 @@ describe('Observable.prototype.publishBehavior', () => {
     expect(results).to.deep.equal([]);
     done();
   });
+
+  type('should infer the type', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.ConnectableObservable<number> = source.publishBehavior(0);
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    // TODO: https://github.com/ReactiveX/rxjs/issues/2972
+    const result: Rx.ConnectableObservable<number> = Rx.operators.publishBehavior(0)(source);
+    /* tslint:enable:no-unused-variable */
+  });
 });

spec/operators/publishLast-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;
@@ -257,4 +258,19 @@ describe('Observable.prototype.publishLast', () => {
     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.ConnectableObservable<number> = source.publishLast();
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    // TODO: https://github.com/ReactiveX/rxjs/issues/2972
+    const result: Rx.ConnectableObservable<number> = Rx.operators.publishLast()(source);
+    /* tslint:enable:no-unused-variable */
+  });
 });

spec/operators/publishReplay-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;
@@ -480,4 +481,47 @@ describe('Observable.prototype.publishReplay', () => {
     expectObservable(published).toBe(expected, undefined, error);
     expectSubscriptions(source.subscriptions).toBe(sourceSubs);
   });
+
+  type('should infer the type', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.ConnectableObservable<number> = source.publishReplay(1);
+    /* 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.publishReplay(1, undefined, 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.publishReplay(1, undefined, s => s.map(x => x + '!'));
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    // TODO: https://github.com/ReactiveX/rxjs/issues/2972
+    const result: Rx.ConnectableObservable<number> = Rx.operators.publishReplay(1)(source);
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator with a selector', () => {
+    /* tslint:disable:no-unused-variable */
+    const source = Rx.Observable.of<number>(1, 2, 3);
+    const result: Rx.Observable<number> = source.pipe(Rx.operators.publishReplay(1, undefined, s => s.map(x => x)));
+    /* tslint:enable:no-unused-variable */
+  });
+
+  type('should infer the type for the pipeable operator 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.pipe(Rx.operators.publishReplay(1, undefined, s => s.map(x => x + '!')));
+    /* tslint:enable:no-unused-variable */
+  });
 });

src/operators/multicast.ts

@@ -3,10 +3,10 @@ import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
 import { ConnectableObservable, connectableObservableDescriptor } from '../observable/ConnectableObservable';
-import { FactoryOrValue, MonoTypeOperatorFunction, OperatorFunction } from '../interfaces';
+import { FactoryOrValue, MonoTypeOperatorFunction, OperatorFunction, UnaryFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
-export function multicast<T>(subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): MonoTypeOperatorFunction<T>;
+export function multicast<T>(subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): UnaryFunction<Observable<T>, ConnectableObservable<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 */

src/operators/publish.ts

@@ -1,9 +1,11 @@
+import { Observable } from '../Observable';
 import { Subject } from '../Subject';
 import { multicast } from './multicast';
-import { MonoTypeOperatorFunction, OperatorFunction } from '../interfaces';
+import { ConnectableObservable } from '../observable/ConnectableObservable';
+import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
-export function publish<T>(): MonoTypeOperatorFunction<T>;
+export function publish<T>(): UnaryFunction<Observable<T>, ConnectableObservable<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 */

src/operators/publishLast.ts

@@ -1,9 +1,9 @@
 import { Observable } from '../Observable';
 import { AsyncSubject } from '../AsyncSubject';
 import { multicast } from './multicast';
-import { OperatorFunction } from '../interfaces';
+import { ConnectableObservable } from '../observable/ConnectableObservable';
+import { UnaryFunction } from '../interfaces';
 
-//TODO(benlesh): specify that the second type is actually a ConnectableObservable
-export function publishLast<T>(): OperatorFunction<T, T> {
+export function publishLast<T>(): UnaryFunction<Observable<T>, ConnectableObservable<T>> {
   return (source: Observable<T>) => multicast(new AsyncSubject<T>())(source);
 }