1import { Observable } from '../Observable'; 2import { Subject } from '../Subject'; 3import { multicast } from './multicast'; 4import { ConnectableObservable } from '../observable/ConnectableObservable'; 5import { MonoTypeOperatorFunction, OperatorFunction, UnaryFunction, ObservableInput, ObservedValueOf } from '../types'; 6 7/* tslint:disable:max-line-length */ 8export function publish<T>(): UnaryFunction<Observable<T>, ConnectableObservable<T>>; 9export function publish<T, O extends ObservableInput<any>>(selector: (shared: Observable<T>) => O): OperatorFunction<T, ObservedValueOf<O>>; 10export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>; 11/* tslint:enable:max-line-length */ 12 13/** 14 * Returns a ConnectableObservable, which is a variety of Observable that waits until its connect method is called 15 * before it begins emitting items to those Observers that have subscribed to it. 16 * 17 * <span class="informal">Makes a cold Observable hot</span> 18 * 19 * ![](publish.png) 20 * 21 * ## Examples 22 * Make source$ hot by applying publish operator, then merge each inner observable into a single one 23 * and subscribe. 24 * ```ts 25 * import { of, zip, interval, merge } from "rxjs"; 26 * import { map, publish, tap } from "rxjs/operators"; 27 * 28 * const source$ = zip(interval(2000), of(1, 2, 3, 4, 5, 6, 7, 8, 9)).pipe( 29 * map(values => values[1]) 30 * ); 31 * 32 * source$ 33 * .pipe( 34 * publish(multicasted$ => 35 * merge( 36 * multicasted$.pipe(tap(x => console.log('Stream 1:', x))), 37 * multicasted$.pipe(tap(x => console.log('Stream 2:', x))), 38 * multicasted$.pipe(tap(x => console.log('Stream 3:', x))), 39 * ) 40 * ) 41 * ) 42 * .subscribe(); 43 * 44 * // Results every two seconds 45 * // Stream 1: 1 46 * // Stream 2: 1 47 * // Stream 3: 1 48 * // ... 49 * // Stream 1: 9 50 * // Stream 2: 9 51 * // Stream 3: 9 52 * ``` 53 * 54 * @param {Function} [selector] - Optional selector function which can use the multicasted source sequence as many times 55 * as needed, without causing multiple subscriptions to the source sequence. 56 * Subscribers to the given source will receive all notifications of the source from the time of the subscription on. 57 * @return A ConnectableObservable that upon connection causes the source Observable to emit items to its Observers. 58 * @method publish 59 * @owner Observable 60 * 61 * 62 */ 63export function publish<T, R>(selector?: OperatorFunction<T, R>): MonoTypeOperatorFunction<T> | OperatorFunction<T, R> { 64 return selector ? 65 multicast(() => new Subject<T>(), selector) : 66 multicast(new Subject<T>()); 67} 68