In RxJS, why can shareReplay(50) keep a chat room's socket stream open after every subscriber has left, and how do you fix it?
answer
- what happens at zero subscribers
- refCount defaults to false
- cache versus live source
- share's reset options
- a grace period before teardown
basics
~20 sshareReplay's refCount defaults to false, so when the subscriber count drops to zero its inner ReplaySubject stays subscribed to the source; for a never-completing source use shareReplay({ bufferSize, refCount: true }) or share with explicit reset options.
solid answer
~40 s`shareReplay` is `share` with a `ReplaySubject` connector and fixed reset rules: reset on error, **not** on completion, and on reaching zero subscribers only if `refCount` is `true` — which defaults to `false`. With the default, the first subscriber opens the socket, and when every component that used it is destroyed the source subscription stays alive: the socket stays open and messages keep filling the buffer with nobody listening. For a one-shot request that completes this is exactly what you want — a cache. For a live source, write `shareReplay({ bufferSize: 50, refCount: true })`: at zero subscribers the source is unsubscribed, and the next subscriber starts a fresh connection with an empty buffer. If rapid leave-and-return would thrash the connection, `share` with `resetOnRefCountZero: () => timer(10_000)` adds a grace period.
code
ts · 14 linesimport { Observable, shareReplay, tap } from 'rxjs';
declare const socket$: Observable<string>;
const messages$ = socket$.pipe(
tap({
subscribe: () => console.log('source subscribed'),
finalize: () => console.log('source released'),
}),
shareReplay({ bufferSize: 50, refCount: true }),
);
const sub = messages$.subscribe();
sub.unsubscribe(); // source released (never logged without refCount: true)go deeper
Recall that shareReplay runs a source once for many subscribers and replays recent values to late ones.
Explain share's connector and its three reset options, and that shareReplay is share with a ReplaySubject, no reset on completion and refCount defaulting to false.
Diagnose a source kept alive after its consumers are gone, pick refCount or a delayed reset for live sources, and keep the default only for completing one-shot results.
Decide where sharing lives — at the source owner or at each consumer — so connection lifetime, caching and teardown are owned in one place rather than per call site.
## What sharing operators do A plain observable is **unicast**: each subscription runs the producer again, so three components subscribing to a socket stream open three sockets. The sharing operators put a **subject** between the source and its subscribers so that one source subscription feeds everyone. In RxJS 7 there are two: - **`share(config?)`** — the configurable one. Its `connector` option creates the subject (a plain `Subject` by default) and three reset options decide when that subject and its source subscription are thrown away. - **`shareReplay(bufferSize?, windowTime?)`** or **`shareReplay({ bufferSize, windowTime, refCount })`** — a thin wrapper: `share` with a `ReplaySubject` connector and fixed reset rules. ## The reset rules `share` can **reset** — unsubscribe from the source and discard its subject, so the next subscriber starts over — at three moments: | Option | `share()` default | `shareReplay` | |---|---|---| | `resetOnError` | `true` | `true` | | `resetOnComplete` | `true` | `false` — a completed source stays cached | | `resetOnRefCountZero` | `true` | the `refCount` setting, default `false` | Each option also accepts a **function returning an observable** instead of a boolean; the reset then happens when that observable emits, which gives a delayed or conditional reset. ## Why the socket stays open 1. The first chat component subscribes to `messages$ = socket$.pipe(shareReplay(50))`. `shareReplay` creates a `ReplaySubject(50)` and subscribes it to `socket$`: the connection opens. 2. Other components subscribe; they share the same subject and get up to 50 buffered messages. 3. The user leaves the room; every component unsubscribes. The subscriber count drops to **zero**. 4. Because `refCount` is `false`, nothing resets: the `ReplaySubject` stays subscribed to `socket$`. The socket stays open, messages keep arriving and are buffered for nobody. 5. Returning to the room re-attaches to the same subject — convenient, but the connection was never released. This default is deliberate: `shareReplay` is often used to **cache** an expensive or one-shot result. For a request that emits once and completes, `resetOnComplete: false` keeps the response and replays it to every later subscriber without re-sending the request — and since the source has completed, there is nothing left running to leak. ## Fixes ```ts import { Observable, ReplaySubject, share, shareReplay, timer } from 'rxjs'; interface Message { from: string; text: string } declare const socket$: Observable<Message>; // never completes on its own // Leaks: the socket stays open with zero subscribers const leaky$ = socket$.pipe(shareReplay(50)); // Closes the socket when the last subscriber leaves const messages$ = socket$.pipe(shareReplay({ bufferSize: 50, refCount: true })); // Keeps it open for 10 s after the last subscriber leaves, then closes it const withGrace$ = socket$.pipe( share({ connector: () => new ReplaySubject<Message>(50), resetOnError: true, resetOnComplete: false, resetOnRefCountZero: () => timer(10_000), }), ); ``` - **`refCount: true`** makes `shareReplay` unsubscribe from the source when the count hits zero. The price: the next subscriber gets a **new** `ReplaySubject`, so the old buffer is gone and the source is subscribed again. - **A delayed reset** via `share` absorbs quick leave-and-return navigation: if someone subscribes again before the timer fires, the pending reset is cancelled and the connection survives. - **Bound the buffer.** `shareReplay()` with no arguments uses an infinite `bufferSize`, which is a second leak on a live source. ## How to spot it in a running app - The network panel shows a socket or poll still active on a screen that no longer uses it. - Logging `subscribe` and `finalize` with `tap` just above the sharing operator shows a source subscription that never finalizes. - Memory grows with the number of messages received, not with what is on screen. ## Choosing - **Live source, no need for history:** `share()` — resets on everything, no replay. - **Live source, late subscribers need the latest values:** `shareReplay({ bufferSize: n, refCount: true })`, or `share` with a `ReplaySubject` connector and a delayed reset. - **One-shot result to cache for the session:** `shareReplay(1)` with the default — it completes, stays cached, and holds nothing open.
- Does refCount: true stop shareReplay from caching a completed HTTP response?No. `refCount` only decides what happens when the subscriber count drops to zero **while the source is still running**. `shareReplay` never resets on completion, so a completed response stays cached and is replayed to later subscribers whether `refCount` is `true` or `false`. An error, by contrast, always resets, so the next subscriber retries the request.
- How do share() and shareReplay() differ for a subscriber that arrives late?`share()` uses a plain `Subject`, so a late subscriber sees only values emitted after it joined. `shareReplay(n)` uses a `ReplaySubject(n)`, so a late subscriber first receives up to `n` buffered values. `share()` also resets on completion, so subscribing after completion runs the source again; `shareReplay` keeps the completed result.
saying these in an interview costs you the question
- shareReplay unsubscribes from its source when the last subscriber leaves, by default
- refCount: true makes shareReplay re-send a request that has already completed
- shareReplay() with no arguments replays only the latest value
- share() replays the last value to late subscribers
- With refCount: true, a returning subscriber still gets the old buffered values