shareReplay

Share source and replay specified number of emissions on subscription.

shareReplay<T>(configOrBufferSize?: number | ShareReplayConfig, windowTime?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>

Parameters

configOrBufferSize	Optional. Default is undefined. Type: number | ShareReplayConfig.
windowTime	Optional. Default is undefined. Maximum time length of the replay buffer in milliseconds.
scheduler	Optional. Default is undefined. Scheduler where connected observers within the selector function will be invoked on.

Returns

MonoTypeOperatorFunction<T>: A function that returns an Observable sequence that contains the elements of a sequence produced by multicasting the source sequence within a selector function.

Description

This operator is a specialization of replay that connects to a source observable and multicasts through a ReplaySubject constructed with the specified arguments. A successfully completed source will stay cached in the shareReplayed observable forever, but an errored source can be retried.

Why use shareReplay?

You generally want to use shareReplay when you have side-effects or taxing computations that you do not wish to be executed amongst multiple subscribers. It may also be valuable in situations where you know you will have late subscribers to a stream that need access to previously emitted values. This ability to replay values on subscription is what differentiates share and shareReplay.

Reference counting

As of RXJS version 6.4.0 a new overload signature was added to allow for manual control over what happens when the operators internal reference counter drops to zero. If refCount is true, the source will be unsubscribed from once the reference count drops to zero, i.e. the inner ReplaySubject will be unsubscribed. All new subscribers will receive value emissions from a new ReplaySubject which in turn will cause a new subscription to the source observable. If refCount is false on the other hand, the source will not be unsubscribed meaning that the inner ReplaySubject will still be subscribed to the source (and potentially run for ever).

Example

import { interval } from 'rxjs';
import { shareReplay, take } from 'rxjs/operators';

const obs$ = interval(1000);
const shared$ = obs$.pipe(
  take(4),
  shareReplay(3)
);
shared$.subscribe(x => console.log('sub A: ', x));
shared$.subscribe(y => console.log('sub B: ', y));

Example for refCount usage

import { interval, Observable, defer } from 'rxjs';
import { shareReplay, take, tap, finalize } from 'rxjs/operators';

const log = <T>(source: Observable<T>, name: string) => defer(() => {
  console.log(`${name}: subscribed`);
  return source.pipe(
    tap({
      next: value => console.log(`${name}: ${value}`),
      complete: () => console.log(`${name}: complete`)
    }),
    finalize(() => console.log(`${name}: unsubscribed`))
  );
});

const obs$ = log(interval(1000), 'source');

const shared$ = log(obs$.pipe(
  shareReplay({bufferSize: 1, refCount: true }),
  take(2),
), 'shared');

shared$.subscribe(x => console.log('sub A: ', x));
shared$.subscribe(y => console.log('sub B: ', y));

// PRINTS:
// shared: subscribed <-- reference count = 1
// source: subscribed
// shared: subscribed <-- reference count = 2
// source: 0
// shared: 0
// sub A: 0
// shared: 0
// sub B: 0
// source: 1
// shared: 1
// sub A: 1
// shared: complete <-- take(2) completes the subscription for sub A
// shared: unsubscribed <-- reference count = 1
// shared: 1
// sub B: 1
// shared: complete <-- take(2) completes the subscription for sub B
// shared: unsubscribed <-- reference count = 0
// source: unsubscribed <-- replaySubject unsubscribes from source observable because the reference count dropped to 0 and refCount is true

// In case of refCount being false, the unsubscribe is never called on the source and the source would keep on emitting, even if no subscribers
// are listening.
// source: 2
// source: 3
// source: 4
// ...

See Also

• publish
• share
• publishReplay