RxJS and Angular Reactive Patterns — Session Recap

RxJS and Angular Reactive Patterns — Session Recap

---

Starting point

No prior RxJS knowledge. Strong OOP background in C#/.NET. Goal: build genuine understanding of RxJS from first principles, then cover operators, Subjects, Signals, and common Angular patterns. Scoped specifically to defending existing Angular work in a backend-focused senior interview — not a frontend role.

---

What we did

Spent the majority of the session on functional programming fundamentals before touching any RxJS API. The OOP mental model kept breaking the functional one — had to strip everything back to plain functions before the Observable concept could land. Once it did, operators, Subjects, and Signals followed relatively quickly.

---

Concepts covered

---

Functional programming — the foundation everything else depends on

The entire session broke down repeatedly until this was solid. The core insight: in JavaScript and C#, functions are values. You can store them in variables, pass them as arguments, and return them from other functions. You don't have to call them immediately.

// A function that takes a number
void Run(Action<int> fn)
{
    fn(1);
    fn(2);
    fn(3);
}

Run(val => Console.WriteLine(val)); // prints 1, 2, 3

fn is a parameter. When you call Run(val => Console.WriteLine(val)), your lambda becomes fn. Calling fn(1) is calling Console.WriteLine(1). No magic. Just substitution.

This is a callback. A function you write that someone else calls. You don't call it yourself. Angular does this with (click)="onButtonClick()" — you wrote onButtonClick, Angular calls it when the event fires.

---

Observable — a stored producer function

Once the callback concept landed, Observable was just this:

// Stripped to bare bones — no class, no API
function startStream(handler) {
  handler(1);
  handler(2);
  handler(3);
}

startStream(val => console.log(val)); // 1, 2, 3

startStream takes your handler. Calls it with values. That is the entire Observable mental model.

new Observable(fn) stores fn. Nothing runs. .subscribe(handler) calls fn(handler). Now it runs.

const stream$ = new Observable(fn => {
  fn(1);
  fn(2);
  fn(3);
});

// Nothing has run yet

stream$.subscribe(val => console.log(val)); // NOW it runs — prints 1, 2, 3

Why push not pull: With IEnumerable you call MoveNext() to pull values — you control timing. With Observable, the producer calls your handler whenever it decides — immediately, after 1 second, on user input, on HTTP response. You react. You don't ask.

Cold Observable: Producer function runs once per subscription. Two subscribers = producer runs twice, independently.

stream$.subscribe(val => console.log('A:', val)); // runs producer once
stream$.subscribe(val => console.log('B:', val)); // runs producer again, independently
// A:1, A:2, A:3, B:1, B:2, B:3

---

Subscription — the cancel handle

.subscribe() returns a Subscription. The Subscription is not the value. It is the handle to cancel.

const sub = stream$.subscribe(val => console.log(val));
// sub is NOT the value — values go to your handler
// sub is the cancel handle

sub.unsubscribe(); // stop receiving values

When to unsubscribe:

• Finite Observables (HTTP calls) — complete themselves. No manual unsubscribe needed.
• Infinite Observables (interval, fromEvent, Subject streams) — never complete. Must unsubscribe in ngOnDestroy or use takeUntilDestroyed. Without this — component destroys, subscription keeps running, memory leak.

// Manual way
export class MyComponent implements OnInit, OnDestroy {
  private sub: Subscription;

  ngOnInit() {
    this.sub = someInfiniteStream$.subscribe(...);
  }

  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

// Modern way — takeUntilDestroyed replaces ngOnDestroy + unsubscribe
export class MyComponent {
  private destroyRef = inject(DestroyRef);

  ngOnInit() {
    someInfiniteStream$.pipe(
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(...);
  }
}

DestroyRef is a reference to the current component's destroy lifecycle. takeUntilDestroyed hooks into it. When Angular destroys the component — Observable completes automatically.

For root services (providedIn: 'root') — service lives for app lifetime, DestroyRef never fires. Either let the stream live forever (fine for global streams) or pass DestroyRef from the component into the service method.

// Component owns lifecycle, passes it to service
ngOnInit() {
  this.stockService.startPolling(this.destroyRef);
}

// Service uses it
startPolling(destroyRef: DestroyRef) {
  interval(3000).pipe(
    switchMap(() => this.http.get('/api/stock-price')),
    takeUntilDestroyed(destroyRef)
  ).subscribe(data => this.currentPrice = data.price);
}

---

pipe and operators — chaining transformations

An operator is a function that takes an Observable and returns a new Observable. pipe applies them left to right. Each operator wraps the previous one. Nothing executes until subscribe.

The LINQ analogy: .Where() and .Select() wrap the enumerable and return new enumerables. Nothing executes until foreach. Same mechanic — different data type (Observable instead of IEnumerable), push instead of pull.

// map — transform each value. Value in, value out.
stream$.pipe(
  map(val => val * 2)
).subscribe(val => console.log(val)); // 2, 4, 6

// filter — drop values that don't pass
stream$.pipe(
  filter(val => val > 2)
).subscribe(val => console.log(val)); // 3 only

// order matters — each operator receives output of the previous
stream$.pipe(
  filter(val => val > 2), // 1 and 2 dropped, 3 passes
  map(val => val * 2)     // 3 → 6
).subscribe(val => console.log(val)); // 6

stream$.pipe(
  map(val => val * 2),    // 1→2, 2→4, 3→6
  filter(val => val > 2)  // 2 passes, 4 passes, 6 passes
).subscribe(val => console.log(val)); // 2, 4, 6

---

Real Angular search field — all operators working together

export class SearchComponent {
  results = [];
  searchControl = new FormControl('');

  constructor(private http: HttpClient) {}

  ngOnInit() {
    this.searchControl.valueChanges.pipe(
      debounceTime(300),               // wait 300ms after user stops typing
      filter(val => val.length >= 3),  // ignore less than 3 characters
      switchMap(val =>                 // cancel previous call, start new one
        this.http.get(`/api/search?q=${val}`)
      )
    ).subscribe(results => {
      this.results = results;
    });
  }
}

valueChanges — Observable that emits on every keystroke. debounceTime(300) — swallows rapid keystrokes, only lets one through after 300ms of silence. filter(val => val.length >= 3) — short values never reach the HTTP call. switchMap — cancels previous in-flight HTTP call when new value arrives. You always get results for what the user is currently typing, not stale results from 3 keystrokes ago.

---

The four flattening operators — what each does and when to use it

All four do the same thing conceptually: for each value, start a new inner Observable (usually an HTTP call). They differ in how they handle a new value arriving while the previous inner Observable is still running.

// switchMap — CANCEL previous, start new
// Use: search field — only latest result matters
searchControl.valueChanges.pipe(
  debounceTime(300),
  switchMap(val => this.http.get(`/search?q=${val}`))
).subscribe(results => this.results = results);

// concatMap — QUEUE — wait for previous to finish, then start next
// Use: file uploads where order matters
fileUploads$.pipe(
  concatMap(file => this.uploadService.upload(file))
).subscribe(result => console.log('uploaded:', result));

// exhaustMap — IGNORE new until current finishes
// Use: payment button — first click submits, all subsequent clicks ignored until done
paymentClicks$.pipe(
  exhaustMap(() => this.paymentService.submit())
).subscribe(result => console.log('payment result:', result));

// mergeMap — PARALLEL — run all simultaneously, results arrive as each finishes
// Use: loading multiple user profiles where order doesn't matter
userIds$.pipe(
  mergeMap(id => this.http.get(`/users/${id}`))
).subscribe(user => console.log(user));

Operator	New value arrives while previous running	Use when
switchMap	Cancel previous, start new	Only latest matters
concatMap	Queue — wait for previous	Order matters
exhaustMap	Ignore new	Prevent double submission
mergeMap	Run both in parallel	Order doesn't matter, go fast

mergeMap result order: Results arrive one by one as each inner Observable completes — whichever HTTP call finishes first comes out first. No guaranteed order.

The rule: If your transformation returns a plain value → map. If it returns an Observable (HTTP call, any async operation) → one of the four above.

---

forkJoin — run multiple, wait for all

Not a pipe operator. Used when you need results from multiple independent Observables before doing anything.

// Page load — need profile, settings, permissions before rendering
forkJoin([
  this.http.get('/user/profile'),
  this.http.get('/user/settings'),
  this.http.get('/user/permissions')
]).subscribe(([profile, settings, permissions]) => {
  this.profile = profile;
  this.settings = settings;
  this.permissions = permissions;
  this.ready = true;
});

All three fire simultaneously. Nothing emits until all three complete. If any one errors or never completes — forkJoin never emits.

forkJoin vs mergeMap: forkJoin — fixed list of Observables known upfront, need all results together. mergeMap — values arrive dynamically, results processed as they come.

---

Subject — an Observable you can push into

Plain Observable is sealed. You cannot push values into it from outside. Subject breaks that restriction.

const subject = new Subject<number>();

subject.subscribe(val => console.log('A:', val));
subject.subscribe(val => console.log('B:', val));

subject.next(1); // A: 1, B: 1
subject.next(2); // A: 2, B: 2

Both subscribers get the same value at the same time. One execution shared across all subscribers — unlike plain Observable where producer runs per subscriber.

Pattern — service as event bus:

@Injectable({ providedIn: 'root' })
export class OrderService {
  private orderDeleted = new Subject<string>();
  orderDeleted$ = this.orderDeleted.asObservable(); // consumers subscribe only, cannot push

  deleteOrder(orderId: string) {
    // delete logic here
    this.orderDeleted.next(orderId); // push event to all subscribers
  }
}

// Component that triggers deletion
export class OrderListComponent {
  constructor(private orderService: OrderService) {}

  onDeleteClick(orderId: string) {
    this.orderService.deleteOrder(orderId);
  }
}

// Component that reacts to deletion
export class NotificationComponent implements OnInit, OnDestroy {
  private sub: Subscription;

  constructor(private orderService: OrderService) {}

  ngOnInit() {
    this.sub = this.orderService.orderDeleted$.subscribe(orderId => {
      console.log(`Order ${orderId} deleted`);
    });
  }

  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

asObservable() strips .next() from the exposed stream. Consumers can subscribe but cannot push. Same principle as a private setter in C# — encapsulation.

Late subscriber with plain Subject: Values emitted before subscribe are gone forever. If NotificationComponent loads 5 seconds after deleteOrder was called — it gets nothing.

---

BehaviorSubject — Subject with memory of last value

// Plain Subject — late subscriber misses everything
const subject = new Subject<number>();
subject.next(1);
subject.next(2);
subject.subscribe(val => console.log(val)); // subscribes now
subject.next(3); // prints 3 only — missed 1 and 2

// BehaviorSubject — late subscriber gets last value immediately
const bSubject = new BehaviorSubject<number>(0); // initial value required
bSubject.next(1);
bSubject.next(2);
bSubject.subscribe(val => console.log(val)); // immediately prints 2, then...
bSubject.next(3); // prints 3
// total output: 2, 3

Read current value without subscribing:

console.log(bSubject.value); // 2 — only BehaviorSubject has .value

Subject vs BehaviorSubject in the same service:

@Injectable({ providedIn: 'root' })
export class OrderService {
  private orderDeleted = new Subject<string>();
  orderDeleted$ = this.orderDeleted.asObservable(); // event — something happened

  private deletedCount = new BehaviorSubject<number>(0);
  deletedCount$ = this.deletedCount.asObservable(); // state — current count

  deleteOrder(orderId: string) {
    this.orderDeleted.next(orderId);
    this.deletedCount.next(this.deletedCount.value + 1);
  }
}

// Badge loads at any time — immediately gets current count
export class BadgeComponent implements OnInit, OnDestroy {
  count = 0;
  private sub: Subscription;

  constructor(private orderService: OrderService) {}

  ngOnInit() {
    this.sub = this.orderService.deletedCount$.subscribe(count => {
      this.count = count; // gets current value immediately on subscribe
    });
  }

  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

The rule: "Is there a current state to know?" → BehaviorSubject "Did something happen?" → Subject

Examples:

• Current user, cart contents, selected theme, feature flags → BehaviorSubject
• Order deleted, error occurred, notification triggered → Subject

---

ReplaySubject — Subject with memory of last N values

const replay = new ReplaySubject<string>(3); // remember last 3

replay.next('msg1');
replay.next('msg2');
replay.next('msg3');
replay.next('msg4');

replay.subscribe(val => console.log(val));
// immediately prints msg2, msg3, msg4 — last 3

Use for chat history, audit logs, anything where late subscribers need recent history not just latest.

---

Signals — reactive state without Observables

Signals are synchronous tracked values. Angular knows exactly what changed and only updates the parts of the UI that depend on it.

export class CartComponent {
  // signal — a tracked value
  items = signal<string[]>([]);

  // computed — derived from signal, auto-updates when items changes
  itemCount = computed(() => this.items().length);
  isEmpty = computed(() => this.items().length === 0);

  // effect — side effect, runs when any signal it reads changes
  constructor() {
    effect(() => {
      console.log('item count changed to', this.itemCount());
      // runs automatically whenever itemCount changes
      // tracks all signals read inside — itemCount depends on items
      // so this fires when items changes
    });
  }

  addItem(item: string) {
    // update — when new value depends on old value
    this.items.update(current => [...current, item]);
  }

  clearCart() {
    // set — when you know the new value directly
    this.items.set([]);
  }

  setTheme(theme: string) {
    this.theme.set(theme);
  }
}

set vs update:

count.set(5);              // I know the new value — 5
count.update(v => v + 1);  // new value depends on old value

computed vs effect:

// computed — I need a derived value
itemCount = computed(() => this.items().length); // value I can read

// effect — I need to do something when a value changes
effect(() => localStorage.setItem('theme', this.theme())); // action, not a value

You cannot call .set() inside an effect. That is what computed is for. Effect is for side effects only — logging, localStorage, HTTP calls.

One effect tracks multiple signals — it re-runs when any signal it reads changes:

effect(() => {
  console.log(`${this.firstName()} ${this.lastName()}`);
  // fires when firstName changes OR when lastName changes
});

Signals vs Subjects: Signals replace BehaviorSubject for component and service state. Cleaner, no subscribe, no unsubscribe, template reads directly. Subjects still needed for events, HTTP streams, WebSocket — anything genuinely async and push-based. toSignal() converts an Observable to a Signal where you need to bridge the two.

---

setInterval polling problem and the RxJS fix

// The problem — plain setInterval with async
let currentPrice = 0;

setInterval(async () => {
  const response = await fetch('/api/stock-price');
  const data = await response.json();
  currentPrice = data.price; // whoever finishes last wins
}, 3000);

Problem 1 — race condition: Interval fires every 3s. If fetch takes 4s, calls overlap. Call 1 started at 0s, call 3 started at 6s, call 3 finishes at 9s, call 1 finishes at 10s — stale data from 0s overwrites fresh data from 6s. Last to finish wins regardless of which is latest.

Problem 2 — no cleanup: No way to stop the interval or cancel in-flight fetches when component destroys.

await vs .then makes no difference — both are async, both have the same problem. The issue is multiple in-flight requests writing to the same variable with no coordination.

Partial fix — recursive setTimeout:

let stopped = false;

async function poll() {
  if (stopped) return;
  const response = await fetch('/api/stock-price');
  const data = await response.json();
  currentPrice = data.price;
  setTimeout(poll, 3000); // next call only after this one finishes
}

poll();

function stop() { stopped = true; }

Solves overlap — next fetch only starts after previous completes. Still manual cleanup.

Full fix in Angular — RxJS:

export class StockComponent {
  private destroyRef = inject(DestroyRef);
  currentPrice = 0;

  ngOnInit() {
    interval(3000).pipe(
      switchMap(() => this.http.get<{price: number}>('/api/stock-price')),
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(data => {
      this.currentPrice = data.price;
    });
  }
}

interval(3000) — emits every 3 seconds. switchMap — cancels previous HTTP call before starting new one. No overlap, no stale data. takeUntilDestroyed — stops everything when component destroys. No leak.

---

What we messed up

Naming confusion — callback looked like recursion Using callback as the parameter name inside a function that also accepted something called a callback made it look like the function was calling itself. It was not — just coincidental naming in two separate scopes. Fix was renaming to whenListenIsCalled and then later stripping the class entirely and using plain functions.

Trying to trace function substitutions mentally Attempting to trace fn => fn(x * 2) substitutions step by step broke repeatedly. Resolution: stop tracing internals. Know what each operator does at the behaviour level. You don't trace LINQ internals to use .Where().

Observable wrapping Observable concept Took many attempts to land. The breakthrough was going all the way back to two plain functions with no classes — startStream and doubled. Once those two functions made sense, the class wrapper made sense.

---

Key values and config to remember

Concept	Rule
map input/output	Plain value in, plain value out
switchMap input/output	Plain value in, Observable out — switchMap subscribes for you
Operator order	Left to right — each receives output of previous
Subject	No memory — late subscribers miss values
BehaviorSubject	Requires initial value — .value property available
asObservable()	Strips .next() — read-only for consumers
set vs update	set when you know value, update when it depends on current
computed vs effect	computed = derived value, effect = side effect
Cannot set inside effect	Use computed for derived values instead
takeUntilDestroyed	Inject DestroyRef as field, pass to operator

---

Unanswered questions / things to investigate

• ReplaySubject — covered definition only, no drill
• toSignal() — mentioned but not demonstrated
• How Signals interact with OnPush change detection
• combineLatest and zip — not covered, may come up

---

What's next

1. Angular architecture — component vs service responsibilities, OnPush change detection and why it matters for performance, smart vs dumb components
2. Offline-first patterns — IndexedDB service layer design, background sync queue, how Ports and Adapters applies at the frontend
3. Defend the 26s → 4.4s cold start — walk through exactly what the redundant sequential calls were, what query consolidation means, how to explain it in an interview