RxJS Observables Explained — Internals, Operators and Production Pitfalls

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);

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);

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);

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);