fix(types): support returning union types from multicast projections

Author: benlesh

spec-dtslint/operators/multicast-spec.ts

@@ -17,6 +17,11 @@ it('should be possible to use a selector', () => {
   const r = of(1, 2, 3).pipe(multicast(() => new Subject<number>(), p => of('foo'))); // $ExpectType Observable<string>
 });
 
+it('should support union types', () => {
+  const o = of(1, 2, 3).pipe(multicast(new Subject<number>(), p => Math.random() > 0.5 ? of(123) : of('foo'))); // $ExpectType Observable<string | number>
+  const p = of(1, 2, 3).pipe(multicast(() => new Subject<number>(), p => Math.random() > 0.5 ? of(123) : of('foo'))); // $ExpectType Observable<string | number>
+});
+
 it('should enforce types', () => {
   const p = of(1, 2, 3).pipe(multicast()); // $ExpectError
 });

spec-dtslint/operators/publish-spec.ts

@@ -22,3 +22,7 @@ it('should infer correctly with parameter', () => {
 it('should enforce type on selector', () => {
   const a = of(1, 2, 3).pipe(publish((x: Observable<string>) => x)); // $ExpectError
 });
+
+it('should support union types in selector', () => {
+  const a = of(1, 2, 3).pipe(publish(() => Math.random() > 0.5 ? of(123) : of('test'))); // $ExpectType Observable<string | number>
+});

spec-dtslint/operators/publishReplay-spec.ts

@@ -21,10 +21,18 @@ it('should accept windowTime, bufferSize, selector of OperatorFunction', () => {
   const a = of(1, 2, 3).pipe(publishReplay(1, 1, (x) => of('a'))); // $ExpectType Observable<string>
 });
 
+it('should accept windowTime, bufferSize, selector returning union type', () => {
+  const a = of(1, 2, 3).pipe(publishReplay(1, 1,  () => Math.random() > 0.5 ? of(123) : of('test'))); // $ExpectType Observable<string | number>
+});
+
 it('should accept windowTime, bufferSize, selector  of MonoTypeOperatorFunction', () => {
   const a = of(1, 2, 3).pipe(publishReplay(1, 1, (x) => x)); // $ExpectType Observable<number>
 });
 
+it('should accept windowTime, bufferSize, selector returning union type, and a scheduler', () => {
+  const a = of(1, 2, 3).pipe(publishReplay(1, 1, () => Math.random() > 0.5 ? of(123) : of('test'), asyncScheduler)); // $ExpectType Observable<string | number>
+});
+
 it('should accept windowTime, bufferSize, selector of OperatorFunction, and scheduler', () => {
   const a = of(1, 2, 3).pipe(publishReplay(1, 1, (x) => of('a'), asyncScheduler)); // $ExpectType Observable<string>
 });

src/internal/operators/multicast.ts

@@ -3,15 +3,13 @@ import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
 import { ConnectableObservable, connectableObservableDescriptor } from '../observable/ConnectableObservable';
-import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction } from '../types';
+import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction, ObservedValueOf, ObservableInput } from '../types';
 
 /* tslint:disable:max-line-length */
 export function multicast<T>(subject: Subject<T>): UnaryFunction<Observable<T>, ConnectableObservable<T>>;
-export function multicast<T>(subject: Subject<T>, selector?: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
-export function multicast<T, R>(subject: Subject<T>, selector?: OperatorFunction<T, R>): OperatorFunction<T, R>;
-export function multicast<T>(SubjectFactory: (this: Observable<T>) => 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>;
+export function multicast<T, O extends ObservableInput<any>>(subject: Subject<T>, selector: (shared: Observable<T>) => O): UnaryFunction<Observable<T>, ConnectableObservable<ObservedValueOf<O>>>;
+export function multicast<T>(subjectFactory: (this: Observable<T>) => Subject<T>): UnaryFunction<Observable<T>, ConnectableObservable<T>>;
+export function multicast<T, O extends ObservableInput<any>>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector: (shared: Observable<T>) => O): OperatorFunction<T, ObservedValueOf<O>>;
 /* tslint:enable:max-line-length */
 
 /**

src/internal/operators/publish.ts

@@ -2,11 +2,11 @@ import { Observable } from '../Observable';
 import { Subject } from '../Subject';
 import { multicast } from './multicast';
 import { ConnectableObservable } from '../observable/ConnectableObservable';
-import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction } from '../types';
+import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction, ObservableInput, ObservedValueOf } from '../types';
 
 /* tslint:disable:max-line-length */
 export function publish<T>(): UnaryFunction<Observable<T>, ConnectableObservable<T>>;
-export function publish<T, R>(selector: OperatorFunction<T, R>): OperatorFunction<T, R>;
+export function publish<T, O extends ObservableInput<any>>(selector: (shared: Observable<T>) => O): OperatorFunction<T, ObservedValueOf<O>>;
 export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
 /* tslint:enable:max-line-length */
 

src/internal/operators/publishReplay.ts

@@ -2,12 +2,11 @@ import { Observable } from '../Observable';
 import { ReplaySubject } from '../ReplaySubject';
 import { multicast } from './multicast';
 import { ConnectableObservable } from '../observable/ConnectableObservable';
-import { UnaryFunction, MonoTypeOperatorFunction, OperatorFunction, SchedulerLike } from '../types';
+import { UnaryFunction, MonoTypeOperatorFunction, OperatorFunction, SchedulerLike, ObservableInput, ObservedValueOf } from '../types';
 
 /* tslint:disable:max-line-length */
 export function publishReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
-export function publishReplay<T, R>(bufferSize?: number, windowTime?: number, selector?: OperatorFunction<T, R>, scheduler?: SchedulerLike): OperatorFunction<T, R>;
-export function publishReplay<T>(bufferSize?: number, windowTime?: number, selector?: MonoTypeOperatorFunction<T>, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
+export function publishReplay<T, O extends ObservableInput<any>>(bufferSize?: number, windowTime?: number, selector?: (shared: Observable<T>) => O, scheduler?: SchedulerLike): OperatorFunction<T, ObservedValueOf<O>>;
 /* tslint:enable:max-line-length */
 
 export function publishReplay<T, R>(bufferSize?: number,