connect

function stable

Creates an observable by multicasting the source within a function that allows the developer to define the usage of the multicast prior to connection.

connect<T, O extends ObservableInput<unknown>>(selector: (shared: Observable<T>) => O, config: ConnectConfig<T> = DEFAULT_CONFIG): OperatorFunction<T, ObservedValueOf<O>>

Parameters

selector

A function used to set up the multicast. Gives you a multicast observable that is not yet connected. With that, you're expected to create and return and Observable, that when subscribed to, will utilize the multicast observable. After this function is executed -- and its return value subscribed to -- the the operator will subscribe to the source, and the connection will be made.

config

Optional. Default is DEFAULT_CONFIG.

The configuration object for connect.

Returns

OperatorFunction<T, ObservedValueOf<O>>

Description

This is particularly useful if the observable source you wish to multicast could be synchronous or asynchronous. This sets it apart from share, which, in the case of totally synchronous sources will fail to share a single subscription with multiple consumers, as by the time the subscription to the result of share has returned, if the source is synchronous its internal reference count will jump from 0 to 1 back to 0 and reset.

To use connect, you provide a selector function that will give you a multicast observable that is not yet connected. You then use that multicast observable to create a resulting observable that, when subscribed, will set up your multicast. This is generally, but not always, accomplished with merge.

Note that using a takeUntil inside of connect's selector might mean you were looking to use the takeWhile operator instead.

When you subscribe to the result of connect, the selector function will be called. After the selector function returns, the observable it returns will be subscribed to, then the multicast will be connected to the source.

Example

Sharing a totally synchronous observable

import { defer, of } from 'rxjs';
import { tap, connect } from 'rxjs/operators';

const source$ = defer(() => {
 console.log('subscription started');
 return of(1, 2, 3, 4, 5).pipe(
   tap(n => console.log(`source emitted ${n}`))
 );
});

source$.pipe(
 // Notice in here we're merging 3 subscriptions to `shared$`.
 connect((shared$) => merge(
     shared$.pipe(map(n => `all ${n}`)),
     shared$.pipe(filter(n => n % 2 === 0), map(n => `even ${n}`)),
     shared$.pipe(filter(n => n % 2 === 1), map(n => `odd ${n}`)),
 ))
)
.subscribe(console.log);

// Expected output: (notice only one subscription)
"subscription started"
"source emitted 1"
"all 1"
"odd 1"
"source emitted 2"
"all 2"
"even 2"
"source emitted 3"
"all 3"
"odd 3"
"source emitted 4"
"all 4"
"even 4"
"source emitted 5"
"all 5"
"odd 5"

© 2015–2021 Google, Inc., Netflix, Inc., Microsoft Corp. and contributors.
Code licensed under an Apache-2.0 License. Documentation licensed under CC BY 4.0.
https://rxjs.dev/api/operators/connect