Reactive Programming

Reactive Programming

Table of Contents

• Overview
• When to Use
• Quick Start
• Reference Guides
• Best Practices

Overview

Build responsive applications using reactive streams and observables for handling asynchronous data flows.

When to Use

• Complex async data flows
• Real-time data updates
• Event-driven architectures
• UI state management
• WebSocket/SSE handling
• Combining multiple data sources

Quick Start

Minimal working example:

import {
  Observable,
  Subject,
  BehaviorSubject,
  fromEvent,
  interval,
} from "rxjs";
import {
  map,
  filter,
  debounceTime,
  distinctUntilChanged,
  switchMap,
} from "rxjs/operators";

// Create observable from array
const numbers$ = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

numbers$.subscribe({
  next: (value) => console.log(value),
// ... (see reference guides for full implementation)

Reference Guides

Detailed implementations in the references/ directory:

Best Practices

✅ DO

• Unsubscribe to prevent memory leaks
• Use operators to transform data
• Handle errors properly
• Use shareReplay for expensive operations
• Combine streams when needed
• Test reactive code

❌ DON'T

• Subscribe multiple times to same observable
• Forget to unsubscribe
• Use nested subscriptions
• Ignore error handling
• Make observables stateful

Backpressure Handling

Backpressure Handling

import { Subject } from "rxjs";
import { bufferTime, throttleTime } from "rxjs/operators";

// Buffer events
const events$ = new Subject<string>();

events$
  .pipe(
    bufferTime(1000), // Collect events for 1 second
    filter((buffer) => buffer.length > 0),
  )
  .subscribe((events) => {
    console.log("Batch:", events);
    processBatch(events);
  });

// Throttle events
const clicks$ = fromEvent(button, "click");

clicks$
  .pipe(
    throttleTime(1000), // Only allow one every second
  )
  .subscribe(() => {
    console.log("Click processed");
  });

function processBatch(events: string[]) {
  // Process batch
}

Combining Multiple Streams

Combining Multiple Streams

import { combineLatest, merge, forkJoin, zip } from "rxjs";

// combineLatest - emits when any input emits
const users$ = fetchUsers();
const settings$ = fetchSettings();

combineLatest([users$, settings$]).subscribe(([users, settings]) => {
  console.log("Users:", users);
  console.log("Settings:", settings);
});

// merge - combine multiple observables
const clicks$ = fromEvent(button1, "click");
const hovers$ = fromEvent(button2, "mouseover");

merge(clicks$, hovers$).subscribe((event) => {
  console.log("Event:", event.type);
});

// forkJoin - wait for all to complete (like Promise.all)
forkJoin({
  users: fetchUsers(),
  posts: fetchPosts(),
  comments: fetchComments(),
}).subscribe(({ users, posts, comments }) => {
  console.log("All data loaded:", { users, posts, comments });
});

// zip - combine corresponding values
const names$ = of("Alice", "Bob", "Charlie");
const ages$ = of(25, 30, 35);

zip(names$, ages$).subscribe(([name, age]) => {
  console.log(`${name} is ${age} years old`);
});

Custom Operators

Custom Operators

import { Observable } from "rxjs";

function tapLog<T>(message: string) {
  return (source: Observable<T>) => {
    return new Observable<T>((subscriber) => {
      return source.subscribe({
        next: (value) => {
          console.log(message, value);
          subscriber.next(value);
        },
        error: (err) => subscriber.error(err),
        complete: () => subscriber.complete(),
      });
    });
  };
}

// Usage
source$
  .pipe(
    tapLog("Before map:"),
    map((x) => x * 2),
    tapLog("After map:"),
  )
  .subscribe();

RxJS Basics

RxJS Basics

import {
  Observable,
  Subject,
  BehaviorSubject,
  fromEvent,
  interval,
} from "rxjs";
import {
  map,
  filter,
  debounceTime,
  distinctUntilChanged,
  switchMap,
} from "rxjs/operators";

// Create observable from array
const numbers$ = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

numbers$.subscribe({
  next: (value) => console.log(value),
  error: (err) => console.error(err),
  complete: () => console.log("Done"),
});

// Subject (multicast)
const subject = new Subject<number>();

subject.subscribe((value) => console.log("Sub 1:", value));
subject.subscribe((value) => console.log("Sub 2:", value));

subject.next(1); // Both subscribers receive

// BehaviorSubject (with initial value)
const state$ = new BehaviorSubject({ count: 0 });

state$.subscribe((state) => console.log("State:", state));

state$.next({ count: 1 });
state$.next({ count: 2 });

// Operators
const source$ = interval(1000);

source$
  .pipe(
    map((n) => n * 2),
    filter((n) => n > 5),
    take(5),
  )
  .subscribe((value) => console.log(value));

Search with Debounce

Search with Debounce

import { fromEvent } from "rxjs";
import {
  debounceTime,
  distinctUntilChanged,
  switchMap,
  catchError,
} from "rxjs/operators";
import { of } from "rxjs";

const searchInput = document.querySelector("#search") as HTMLInputElement;

const search$ = fromEvent(searchInput, "input").pipe(
  map((event: Event) => (event.target as HTMLInputElement).value),
  debounceTime(300), // Wait 300ms after typing
  distinctUntilChanged(), // Only if value changed
  switchMap((query) => {
    if (!query) return of([]);

    return fetch(`/api/search?q=${query}`)
      .then((res) => res.json())
      .catch(() => of([]));
  }),
  catchError((error) => {
    console.error("Search error:", error);
    return of([]);
  }),
);

search$.subscribe((results) => {
  console.log("Search results:", results);
  displayResults(results);
});

function displayResults(results: any[]) {
  // Update UI
}

State Management

State Management

import { BehaviorSubject } from "rxjs";
import { map } from "rxjs/operators";

interface AppState {
  user: { id: string; name: string } | null;
  cart: Array<{ id: string; quantity: number }>;
  loading: boolean;
}

class StateManager {
  private state$ = new BehaviorSubject<AppState>({
    user: null,
    cart: [],
    loading: false,
  });

  // Selectors
  user$ = this.state$.pipe(
    map((state) => state.user),
    distinctUntilChanged(),
  );

  cart$ = this.state$.pipe(
    map((state) => state.cart),
    distinctUntilChanged(),
  );

  cartTotal$ = this.cart$.pipe(
    map((cart) => cart.reduce((sum, item) => sum + item.quantity, 0)),
  );

  loading$ = this.state$.pipe(map((state) => state.loading));

  // Actions
  setUser(user: AppState["user"]): void {
    this.state$.next({
      ...this.state$.value,
      user,
    });
  }

  addToCart(item: { id: string; quantity: number }): void {
    const cart = [...this.state$.value.cart];
    const existing = cart.find((i) => i.id === item.id);

    if (existing) {
      existing.quantity += item.quantity;
    } else {
      cart.push(item);
    }

    this.state$.next({
      ...this.state$.value,
      cart,
    });
  }

  setLoading(loading: boolean): void {
    this.state$.next({
      ...this.state$.value,
      loading,
    });
  }

  getState(): AppState {
    return this.state$.value;
  }
}

// Usage
const store = new StateManager();

store.user$.subscribe((user) => {
  console.log("User:", user);
});

store.cartTotal$.subscribe((total) => {
  console.log("Cart items:", total);
});

store.setUser({ id: "123", name: "John" });
store.addToCart({ id: "item1", quantity: 2 });

WebSocket with Reconnection

WebSocket with Reconnection

import { Observable, timer } from "rxjs";
import { retryWhen, tap, delayWhen } from "rxjs/operators";

function createWebSocketObservable(url: string): Observable<any> {
  return new Observable((subscriber) => {
    let ws: WebSocket;

    const connect = () => {
      ws = new WebSocket(url);

      ws.onopen = () => {
        console.log("WebSocket connected");
      };

      ws.onmessage = (event) => {
        try {
          const data = JSON.parse(event.data);
          subscriber.next(data);
        } catch (error) {
          console.error("Parse error:", error);
        }
      };

      ws.onerror = (error) => {
        console.error("WebSocket error:", error);
        subscriber.error(error);
      };

      ws.onclose = () => {
        console.log("WebSocket closed");
        subscriber.error(new Error("Connection closed"));
      };
    };

    connect();

    return () => {
      if (ws) {
        ws.close();
      }
    };
  }).pipe(
    retryWhen((errors) =>
      errors.pipe(
        tap((err) => console.log("Retrying connection...", err)),
        delayWhen((_, i) => timer(Math.min(1000 * Math.pow(2, i), 30000))),
      ),
    ),
  );
}

// Usage
const ws$ = createWebSocketObservable("wss://api.example.com/ws");

ws$.subscribe({
  next: (data) => console.log("Received:", data),
  error: (err) => console.error("Error:", err),
});