RxJS Observables — shareReplay Without refCount Leak

Every modern JavaScript app — whether it's an Angular dashboard, a React data-fetching layer, or a Node.js event pipeline — eventually runs into the same problem: asynchronous data that arrives in bursts, needs to be transformed, combined with other streams, and cancelled gracefully. setTimeout and Promises handle one-shot async well, but they fall apart the moment you need to debounce a search box, retry a failing API call with exponential backoff, or merge a WebSocket stream with an HTTP response. RxJS was built exactly for that world.

RxJS (Reactive Extensions for JavaScript) brings the Observer pattern, the Iterator pattern, and functional programming together into one composable API. At its core, an Observable is a lazy, cancellable, composable data pipeline. Unlike a Promise, it can emit multiple values over time, it doesn't start executing until something subscribes to it, and it can be torn down mid-flight — which is the key to avoiding memory leaks in dynamic UIs.

By the end of this article you'll understand how Observables work under the hood, why cold vs hot matters in production, how the most important operators actually compose, how multicasting prevents redundant network calls, and exactly which mistakes ship bugs to production. You'll also walk away with interview-ready answers that go beyond surface-level definitions.

Why shareReplay Without refCount Leaks Memory

An RxJS Observable is a lazy push-based collection — it doesn't emit until subscribed, and each subscriber gets its own execution unless the Observable is made multicast. shareReplay is a multicast operator that caches the last N emissions and replays them to new subscribers. The core mechanic: it wraps a Subject, subscribes once to the source, and replays the buffer to late subscribers. Without refCount: true, the underlying Subject stays subscribed even after all subscribers unsubscribe. This means the source Observable never completes or errors, and the subscription to the source persists indefinitely. In practice, this creates a permanent reference chain: the source holds resources (timers, HTTP connections, WebSocket listeners) that never release. The key property: shareReplay defaults to refCount: false, meaning the operator keeps the source alive until the Observable itself is garbage collected — which may never happen if the Observable is referenced globally or in a long-lived service. Use shareReplay when you need to replay past values to late subscribers, but always pass refCount: true unless you explicitly need the source to stay alive (e.g., a shared WebSocket that should reconnect). In real systems, forgetting refCount: true is the #1 cause of silent memory leaks in Angular services and long-lived RxJS streams.

How Observables Work Internally — Not Just What They Are

Most tutorials treat Observable as a black box. Let's crack it open. At its simplest, an Observable is a function that accepts an Observer (an object with next, error, and complete callbacks) and returns a teardown function. That's the entire contract. When you call subscribe(), RxJS invokes that producer function and wires up the observer. Nothing happens before that call — that's what 'lazy' means.

This is fundamentally different from a Promise, which starts its executor synchronously the moment you call new Promise(). An Observable defers all work until subscription time, which means you can pass an Observable around, compose it with operators, and store it in a variable without triggering any side effects. That referential transparency is what makes Observables safe to compose.

The teardown function returned by the producer (or set via subscriber.add()) is called when you unsubscribe, or when the Observable completes or errors. This is the foundation of RxJS's memory-safety story — every resource (timers, event listeners, WebSocket connections) must be cleaned up in that teardown. If your custom Observable doesn't return a teardown, you've created a leak.

import { Observable } from 'rxjs';

// Creating an Observable from scratch reveals its internals.
// The function we pass IS the producer — it runs only on subscribe().
const intervalObservable = new Observable((subscriber) => {
  let tickCount = 0;
  console.log('[Producer] Observable execution started');

  // setInterval is the resource we must clean up.
  const intervalId = setInterval(() => {
    tickCount++;
    console.log(`[Producer] Emitting tick #${tickCount}`);
    subscriber.next(tickCount); // Push value to observer

    if (tickCount === 3) {
      clearInterval(intervalId);   // Stop the interval ourselves
      subscriber.complete();        // Signal no more values
    }
  }, 500);

  // Return teardown logic — called if consumer unsubscribes early
  return () => {
    console.log('[Teardown] Interval cleared — no leak');
    clearInterval(intervalId);
  };
});

console.log('[Main] Observable defined — nothing running yet');

const subscription = intervalObservable.subscribe({
  next: (tick) => console.log(`[Observer] Received: ${tick}`),
  error: (err) => console.error(`[Observer] Error: ${err}`),
  complete: () => console.log('[Observer] Stream complete'),
});

// Uncomment to see teardown fire before tick #3:
// setTimeout(() => subscription.unsubscribe(), 800);

Cold vs Hot Observables — The Distinction That Ships Bugs

This is the single most misunderstood concept in RxJS and the root cause of both duplicate API calls and missed WebSocket messages. Understanding it deeply separates senior RxJS engineers from everyone else.

A cold Observable creates its producer fresh for each subscriber. Each subscriber gets the complete sequence from the beginning, with its own independent execution context. The interval example above is cold — two subscribers would each get their own timer. fromFetch(), ajax(), and interval() are cold by default.

A hot Observable shares a single producer among all subscribers. Subscribers only receive values emitted after they subscribe — like a live concert stream. fromEvent() (wrapping a DOM event) is hot because there's one event listener on the element, not one per subscriber.

The danger zone is HTTP requests: if you build a search-as-you-type feature using a cold ajax() Observable and render it in two places, each render triggers a separate HTTP request. The fix is multicasting — turning a cold Observable hot so all subscribers share one execution. shareReplay(1) is the production workaround most Angular devs reach for, but it has its own subtleties around refCounting and memory.

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

// ─── COLD Observable demonstration ─────────────────────────────────────────
const coldTimer$ = new Observable((subscriber) => {
  // Each subscriber gets its own counter starting from 0
  let count = 0;
  const id = setInterval(() => subscriber.next(count++), 300);
  return () => clearInterval(id);
}).pipe(take(3));

console.log('--- COLD: two subscribers get independent streams ---');
coldTimer$.subscribe(v => console.log(`Cold Subscriber A: ${v}`));
setTimeout(() => {
  coldTimer$.subscribe(v => console.log(`Cold Subscriber B: ${v}`));
}, 400); // B starts 400ms late — still gets 0, 1, 2 from its own producer

// ─── HOT Observable via Subject ────────────────────────────────────────────
// A Subject is both an Observable and an Observer — it's the canonical hot source.
const liveScore$ = new Subject();

setTimeout(() => {
  console.log('\n--- HOT: Subject shares one stream ---');

  liveScore$.subscribe(score => console.log(`Fan A sees: ${score}`));

  // Emit the first goal — both fans see it (if subscribed at that moment)
  liveScore$.next('Goal! 1-0');

  // Fan B subscribes after the first goal — they MISS it
  liveScore$.subscribe(score => console.log(`Fan B sees: ${score}`));
  liveScore$.next('Goal! 2-0'); // Only this one is seen by Fan B
  liveScore$.complete();
}, 1200);

// ─── shareReplay: solve the cold HTTP + multiple subscriber problem ─────────
// Simulating an HTTP call that should only fire ONCE
const expensiveApiCall$ = new Observable((subscriber) => {
  console.log('\n[HTTP] Making API call — this should only appear ONCE');
  setTimeout(() => {
    subscriber.next({ user: 'Alice', role: 'admin' });
    subscriber.complete();
  }, 100);
}).pipe(
  shareReplay(1) // Replay last 1 emission to late subscribers; share the single HTTP call
);

setTimeout(() => {
  // Two components subscribe — only one HTTP call fires
  expensiveApiCall$.subscribe(data => console.log('[Component Header] Got:', data));
  expensiveApiCall$.subscribe(data => console.log('[Component Sidebar] Got:', data));
}, 2400);

Operator Internals and Composition — map, switchMap, mergeMap, exhaustMap Compared

Operators are pure functions that take an Observable and return a new Observable. They don't mutate the source — each operator wraps the previous one in a new layer, forming a pipeline. Under the hood, pipe() is just function composition: pipe(opA, opB, opC) is equivalent to opC(opB(opA(source))).

The higher-order mapping operators — switchMap, mergeMap, concatMap, exhaustMap — are where most production bugs live. They all accept a function that maps each emitted value to an inner Observable. The difference is what they do with overlapping inner subscriptions.

switchMap cancels the previous inner Observable when a new outer value arrives. This is perfect for autocomplete — you only care about the response for the latest keystroke. mergeMap subscribes to every inner Observable concurrently, which is useful for parallel requests but can overwhelm a server. concatMap queues them, processing one at a time in order. exhaustMap ignores new outer values while an inner Observable is still active — ideal for a login button that shouldn't fire twice.

Choosing the wrong one causes race conditions (mergeMap for search), dropped requests (exhaustMap for pagination), or stalled queues (concatMap when order doesn't matter but throughput does).

import { fromEvent, of, timer, Subject } from 'rxjs';
import {
  switchMap,
  mergeMap,
  concatMap,
  exhaustMap,
  map,
  delay,
  tap,
  take
} from 'rxjs/operators';

// Simulated async search API — takes longer for 'slow' queries
const fakeSearchApi = (searchTerm) => {
  const responseDelay = searchTerm === 'slow' ? 600 : 200;
  return of(`Results for: "${searchTerm}"`).pipe(
    delay(responseDelay),
    tap(() => console.log(`[API] Response ready for: "${searchTerm}"`)),
  );
};

// ─── switchMap — cancels previous, only latest matters ──────────────────────
const searchInput$ = new Subject(); // simulate keystrokes

console.log('=== switchMap (autocomplete) ===');
const searchResults$ = searchInput$.pipe(
  switchMap((term) => {
    console.log(`[switchMap] New term "${term}" — cancelling previous inner subscription`);
    return fakeSearchApi(term);
  })
);

const sub1 = searchResults$.subscribe(result => console.log('[UI] Showing:', result));

// Simulate rapid typing — 'slow' starts first but 'fast' should win
searchInput$.next('slow');
setTimeout(() => searchInput$.next('fast'), 100); // arrives before 'slow' responds
setTimeout(() => sub1.unsubscribe(), 700);

// ─── exhaustMap — ignores while busy ────────────────────────────────────────
const loginClick$ = new Subject();

setTimeout(() => {
  console.log('\n=== exhaustMap (login button) ===');

  const loginRequest$ = loginClick$.pipe(
    exhaustMap(() => {
      console.log('[Auth] Sending login request...');
      // Simulates a 400ms login round-trip
      return of('Login success').pipe(delay(400));
    })
  );

  const sub2 = loginRequest$.subscribe(result => console.log('[Auth] Result:', result));

  // User double-clicks — second click is ignored while first is in flight
  loginClick$.next('click');
  setTimeout(() => loginClick$.next('double-click'), 100); // ignored
  setTimeout(() => sub2.unsubscribe(), 600);
}, 800);

// ─── concatMap — ordered queue ───────────────────────────────────────────────
setTimeout(() => {
  console.log('\n=== concatMap (ordered file uploads) ===');

  const filesToUpload$ = of('file-1.jpg', 'file-2.jpg', 'file-3.jpg');

  filesToUpload$.pipe(
    concatMap((filename) => {
      console.log(`[Upload] Starting: ${filename}`);
      return of(`${filename} uploaded`).pipe(delay(200));
    })
  ).subscribe(result => console.log('[Upload] Done:', result));
}, 1400);

Production Patterns — Error Handling, Retry and Memory Management

Error handling in RxJS is a trap for the unprepared. When an Observable errors, it terminates — no more values, no recovery. That means if you have a WebSocket stream and it throws, your UI goes silent. The answer is catchError, which intercepts an error and must return a new Observable (including EMPTY to silently swallow it, or throwError to re-throw).

retryWhen and its modern replacement retry({ delay, count }) let you implement exponential backoff — critical for flaky API endpoints. But retry resubscribes to the entire source Observable, which for cold Observables means a fresh HTTP call — exactly what you want. For hot sources, retry can cause confusing behaviour because the source doesn't reset.

For memory management in SPAs, the takeUntilDestroyed() operator (Angular 16+) or the classic takeUntil(destroy$) pattern ensures subscriptions are torn down when a component unmounts. In React with RxJS, cleaning up in useEffect's return function is the equivalent. Forgetting this in a long-lived app with many navigations leads to dozens of stale subscriptions running in the background, causing ghost updates to unmounted components and measurable memory growth you'll only catch in a production heap snapshot.

import { throwError, of, Subject, timer } from 'rxjs';
import {
  catchError,
  retry,
  switchMap,
  takeUntil,
  finalize,
  tap,
  mergeMap,
  delayWhen
} from 'rxjs/operators';

// ─── Exponential backoff retry ───────────────────────────────────────────────
let attemptCount = 0;

const unreliableApi$ = new Subject().pipe(
  // This pattern is used to kick off an Observable manually
);

// Simulates a flaky fetch that succeeds on the 3rd attempt
const fetchWithRetry$ = new (require('rxjs').Observable)((subscriber) => {
  attemptCount++;
  console.log(`[HTTP] Attempt #${attemptCount}`);

  if (attemptCount < 3) {
    subscriber.error(new Error(`Network timeout on attempt ${attemptCount}`));
  } else {
    subscriber.next({ data: 'User profile fetched successfully' });
    subscriber.complete();
  }
}).pipe(
  // retry with exponential backoff — delay doubles each time, max 3 retries
  retry({
    count: 3,
    delay: (error, retryIndex) => {
      const backoffMs = Math.pow(2, retryIndex) * 100; // 200ms, 400ms, 800ms
      console.log(`[Retry] Attempt ${retryIndex} after ${backoffMs}ms — Reason: ${error.message}`);
      return timer(backoffMs);
    },
  }),
  catchError((finalError) => {
    // Only runs if all retries are exhausted
    console.error(`[Error] Giving up after all retries: ${finalError.message}`);
    // Return a fallback value so the stream recovers instead of dying
    return of({ data: 'Cached fallback data', fromCache: true });
  }),
  finalize(() => console.log('[Cleanup] HTTP stream finalized — run cleanup here'))
);

fetchWithRetry$.subscribe({
  next: (result) => console.log('[UI] Displaying:', result),
  complete: () => console.log('[UI] Done'),
});

// ─── takeUntil pattern — prevents memory leaks ──────────────────────────────
setTimeout(() => {
  console.log('\n--- takeUntil (component lifecycle) ---');

  // In a real app this would be a component's ngOnDestroy Subject or React useEffect cleanup
  const componentDestroyed$ = new Subject();

  const liveDataStream$ = new (require('rxjs').interval)(200).pipe(
    tap(tick => console.log(`[Stream] Tick ${tick} — component still alive`)),
    takeUntil(componentDestroyed$) // auto-unsubscribes when destroy$ emits
  );

  liveDataStream$.subscribe({
    next: (tick) => console.log(`[Component] Rendered tick: ${tick}`),
    complete: () => console.log('[Component] Subscription cleaned up — no leak'),
  });

  // Simulate component unmount after 500ms
  setTimeout(() => {
    console.log('[Component] Unmounting...');
    componentDestroyed$.next(true);
    componentDestroyed$.complete();
  }, 500);
}, 1500);

Multicasting Internals — Subject, BehaviorSubject, ReplaySubject, AsyncSubject

When you need to share a single execution among multiple subscribers, you need multicasting. The core mechanism is Subject — a type that is both an Observable and an Observer. You push values into it via next(), and all subscribed observers receive them.

BehaviorSubject extends Subject: it remembers the last emitted value and replays it to new subscribers immediately. This makes it perfect for 'current user' state — when a component initializes, it gets the current user without waiting for a new emission.

ReplaySubject replays a configurable number of past emissions (or all). Use it for caching transient data like search results that you want to show for a few seconds after navigation.

AsyncSubject replays only the last value after the source completes. It's rarely used but perfect for loading a resource that you know will eventually complete.

The key to multicasting correctly is understanding that the Subject is the 'bridge' from cold to hot. You can create a Subject, subscribe your source Observable to it, and then expose the Subject as the hot observable. Operators like share, shareReplay, and publish do this internally. shareReplay is the most common production choice because it also caches the last value for late subscribers.

import { Subject, BehaviorSubject, ReplaySubject, AsyncSubject, of, timer } from 'rxjs';
import { delay, tap } from 'rxjs/operators';

// ─── Subject – basic hot source ───────────────────────────────────────────────
const subject$ = new Subject();
subject$.subscribe(v => console.log(`Subject A: ${v}`));
subject$.next('first');  // A gets it
const subB = subject$.subscribe(v => console.log(`Subject B: ${v}`));
subject$.next('second'); // A and B get it
subject$.next('third');  // A and B get it

// ─── BehaviorSubject – remembers last value ───────────────────────────────────
console.log('\n--- BehaviorSubject ---');
const initial = { user: 'guest' };
const currentUser$ = new BehaviorSubject(initial);

// Late subscriber gets the last value immediately
currentUser$.subscribe(user => console.log(`Sub1: ${user.user}`));

currentUser$.next({ user: 'Alice' });
currentUser$.next({ user: 'Bob' });

// This subscriber will receive Bob immediately upon subscription
currentUser$.subscribe(user => console.log(`Sub2 (late): ${user.user}`));

// ─── ReplaySubject – replays past N emissions ─────────────────────────────────
console.log('\n--- ReplaySubject (buffer 2) ---');
const replay$ = new ReplaySubject(2); // replay last 2 values
replay$.next('a');
replay$.next('b');
replay$.next('c');

// Subscriber gets 'b' and 'c' (the 2 buffered)
replay$.subscribe(v => console.log(`Replay A (late): ${v}`));
replay$.next('d');

// ─── AsyncSubject – emits last value only on complete ─────────────────────────
console.log('\n--- AsyncSubject ---');
const asyncSubject$ = new AsyncSubject();
asyncSubject$.subscribe(v => console.log(`Async Sub1: ${v}`));
asyncSubject$.next('first'); // ignored
asyncSubject$.next('last');   // this is the value kept
asyncSubject$.complete();     // now 'last' is emitted to all subscribers

// Late subscriber also gets 'last'
asyncSubject$.subscribe(v => console.log(`Async Sub2 (late): ${v}`));

The Data Pipeline: Why Your RxJS Code Smells Like Callback Hell

You don't subscribe inside subscribe. I've seen it. A junior wires a user input event, then in the callback manually calls another observable with a nested subscribe. That's not reactive programming — that's callback hell with extra steps.

RxJS data pipelines exist for one reason: to declare transformations declaratively, not imperatively. You take a source observable, pipe it through operators that describe what happens to each value, and then subscribe once at the end. That's it. No nesting. No intermediate subscriptions. No manual cleanup.

The pipe() function composes operators lazily. Each operator returns a new observable that wraps the previous one. When data flows through, each operator transforms or filters it before passing it downstream. This means your subscription handler only sees the final result, not the intermediate noise.

If your observable pipeline has more than one subscription inside it, you've already lost. Refactor into a single pipeline with switchMap, map, or filter.

// io.thecodeforge — javascript tutorial

// Wrong: nested subscribe inside callback
searchInput.valueChanges.subscribe((query) => {
  httpClient.get(`/api/users?q=${query}`).subscribe((users) => {
    renderResults(users);
  });
});

// Right: declarative pipeline, single subscribe
searchInput.valueChanges
  .pipe(
    debounceTime(300),
    filter((q) => q.length >= 2),
    distinctUntilChanged(),
    switchMap((query) => httpClient.get(`/api/users?q=${query}`)),
    catchError((err) => of([]))
  )
  .subscribe({
    next: (users) => renderResults(users),
    error: (err) => console.error('Unhandled error:', err),
    complete: () => console.log('Stream completed')
  });

What Is Reactive Programming? (And Why Your Event Handlers Are Lying to You)

Reactive programming isn't about Observables or RxJS. It's a shift in how you model change over time. Traditional imperative code says: "Do this, then do that, then check this." Reactive programming says: "Here's a stream of events. Here's how to transform it. React as it flows."

You've already done reactive programming without knowing it. Click listeners? That's a stream of click events. Promise chains? That's a stream that emits once. The problem is you're treating each event source as a special case with its own API — addEventListener, then, callbacks. Reactive programming gives you a single abstraction for all of them: the Observable.

The advantage isn't just consistency. It's composition. With imperative code, combining two event sources requires nested callbacks or state variables. With reactive programming, you use combineLatest, merge, or forkJoin. You describe what you want, not how to wire it up. This means less code, fewer bugs, and no state synchronization errors.

Reactive programming also gives you backpressure control, cancellation, and error propagation. Promises fail silently when you forget a catch. Observables let you handle errors exactly where they happen — in the pipeline or at subscription.

// io.thecodeforge — javascript tutorial

// Imperative: manual state tracking
let email = '';
let password = '';
let loading = false;

function onEmailChange(newEmail) {
  email = newEmail;
  checkForm();
}

function onPasswordChange(newPwd) {
  password = newPwd;
  checkForm();
}

function checkForm() {
  if (email.length > 0 && password.length >= 8 && !loading) {
    submitBtn.disabled = false;
  }
}

// Reactive: declarative stream composition
const email$ = fromEvent(emailInput, 'input').pipe(
  map((e) => e.target.value)
);

const password$ = fromEvent(passwordInput, 'input').pipe(
  map((e) => e.target.value)
);

const formValid$ = combineLatest([email$, password$]).pipe(
  map(([email, pwd]) => email.length > 0 && pwd.length >= 8)
);

formValid$.subscribe((isValid) => {
  submitBtn.disabled = !isValid;
});

Why combineLatest With Dynamic Streams Silently Breaks

combineLatest emits a new value whenever any source emits, but only after every source has emitted at least once. When streams are added dynamically (e.g., via array.push into an observable list), the operator doesn't re-evaluate the initial state. New sources may emit later, leaving the combined result in a stale partial state. This causes UI flickers, missing data, or silent failures if downstream operators assume completeness. The fix is to seed each new stream with a startWith value or use forkJoin if you only care about the initial emission. Never assume combineLatest lazily adapts to dynamic source changes—it only tracks what it was given at subscription time. Test your multicasting setup: if a BehaviorSubject feeding combineLatest fires before all sources are ready, your pipeline emits incomplete payloads.

// io.thecodeforge — javascript tutorial

import { combineLatest, Subject } from 'rxjs';
import { startWith } from 'rxjs/operators';

const source1 = new Subject();
const source2 = new Subject();

// Static combineLatest works fine
combineLatest([source1, source2]).subscribe(console.log);
source1.next('A'); // no emit (waiting for source2)
source2.next('B'); // ['A','B']

// Dynamic stream: missing initial value
const dynamic$ = new Subject();
combineLatest([source1, dynamic$.pipe(startWith(null))]).subscribe(console.log);
dynamic$.next('X'); // ['A','X'] or stale if source1 hasn't emitted

Why Subscription.add() Hides Memory Leaks Better Than takeUntil

takeUntil is the standard pattern for unsubscribing when a notifier emits. But it only works if the notifier completes or emits at the right moment. If the notifier itself never fires (e.g., a Subject that never .next()), the subscription stays active forever. subscription.add(childSub) groups subscriptions into a parent container, letting you manage all teardown logic in one place. More importantly, it auto-unsubscribes children if the parent unsubscribes—no reliance on external notifiers. This prevents orphan subscriptions when components are destroyed but the notifier is still alive. Use add for composite subscriptions (e.g., multiple HTTP requests) and takeUntil only when you control the notifier's lifecycle. Never mix both without a finalize operator to force cleanup.

// io.thecodeforge — javascript tutorial

import { interval, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

// Leaky: notifier never fires
const neverFires = new Subject();
interval(1000).pipe(takeUntil(neverFires)).subscribe();
// Leaks forever

// Safe: subscription.add
const bag = new Subject();
const sub = interval(1000).subscribe();
bag.add(sub); // parent manages child
bag.unsubscribe(); // kills sub

// Production: combine both
const notifier = new Subject();
interval(1000).pipe(takeUntil(notifier)).subscribe();
notifier.next(); // clean exit

Why testrx.js Solves Real-World Observable Debugging

When your Angular app’s HTTP stream silently fails or a switchMap drops requests, logging to console won’t reveal the race condition. testrx.js is a lightweight sandbox (npm install testrx) that lets you simulate time with marble diagrams in Node.js without a browser. Unlike full test runners, testrx.js focuses on observables directly: you write marble strings like -a-b-c| to define emissions, then pipe real operators to see exactly when errors or completions fire. The why before how: most devs debug by adding .pipe(tap()) but miss timing bugs because browser DevTools can’t replay async sequences. testrx.js gives you deterministic reproduction. Use it to isolate memory leaks from shareReplay or verify that your retry logic actually waits the correct interval. Always test the observable, not the UI.

// io.thecodeforge — javascript tutorial
const {cold} = require('testrx');
const {map, retryWhen, delay, take} = require('rxjs/operators');

// Simulate a stream that fails twice then works
const source$ = cold('-a-#(e)', {a: 'data', e: new Error('fail')});

source$.pipe(
  map(x => x.toUpperCase()),
  retryWhen(errors => errors.pipe(
    delay(10),  // wait 10ms, simulated
    take(2)     // retry twice
  ))
).subscribe({
  next: v => console.log('Value:', v),
  error: e => console.log('Final error:', e)
});
// Output (simulated): after 30ms, logs 'DATA' only

Why Educative Courses Fill the Production-Reactive Gap

Documentation explains operators; production teaches failure patterns. Educative’s interactive tutorials (like "RxJS Mastery" or "Reactive Patterns Angular") go beyond API docs by having you refactor buggy real-world code — memory leaks from shareReplay, race conditions in combineLatest, or improper unsubscription in React useEffect. The why: most developers learn observables via small examples that never stress-test garbage collection or dynamic stream counts. Educative’s live coding environment lets you run and break code immediately, seeing the memory tab in your browser spike. They also cover web application concerns: handling WebSocket reconnection with retryWhen, debouncing search inputs with switchMap, and canceling stale HTTP requests via AbortController integrated with takeUntil. Use their "RxJS in Production" path to reinforce the patterns from this article — especially why shareReplay without refCount leaks resources across route changes.

// io.thecodeforge — javascript tutorial
const {Subject, interval} = require('rxjs');
const {takeUntil, shareReplay} = require('rxjs/operators');

// Dynamic streams: avoid memory leaks via refCount
const destroy$ = new Subject();
const source$ = interval(1000).pipe(
  takeUntil(destroy$),
  shareReplay({bufferSize: 1, refCount: true})
);

// First subscriber triggers subscription
source$.subscribe(v => console.log('A:', v));
// Second shares window, no new interval
source$.subscribe(v => console.log('B:', v));

destroy$.next(); // Both subscribers complete, underlying stream unsubscribes
console.log('No memory leak — refCount cleaned up');