Reactive Programming Basics

Challenge

Respond automatically to data changes and handle event streams elegantly.

Explanation

Reactive programming treats data and events as streams that flow through your application.

Observable Pattern

class Observable {
  constructor(subscribe) {
    this.subscribe = subscribe;
  }

  static of(...values) {
    return new Observable(observer => {
      values.forEach(value => observer.next(value));
      observer.complete();
    });
  }

  map(fn) {
    return new Observable(observer => {
      return this.subscribe({
        next: (x) => observer.next(fn(x)),
        error: (e) => observer.error(e),
        complete: () => observer.complete()
      });
    });
  }

  filter(predicate) {
    return new Observable(observer => {
      return this.subscribe({
        next: (x) => {
          if (predicate(x)) observer.next(x);
        },
        error: (e) => observer.error(e),
        complete: () => observer.complete()
      });
    });
  }
}

const stream = Observable.of(1, 2, 3, 4, 5);

stream
  .filter(x => x % 2 === 0)
  .map(x => x * 2)
  .subscribe({
    next: (value) => console.log(value),
    error: (err) => console.error(err),
    complete: () => console.log('Done')
  });

Subject Pattern

class Subject {
  constructor() {
    this.observers = [];
  }

  subscribe(observer) {
    this.observers.push(observer);
    return () => {
      this.observers = this.observers.filter(o => o !== observer);
    };
  }

  emit(value) {
    this.observers.forEach(obs => obs.next(value));
  }
}

const button$ = new Subject();

// Multiple subscriptions
button$.subscribe({
  next: (event) => console.log('Logger:', event)
});

button$.subscribe({
  next: (event) => console.log('Analytics:', event)
});

button$.emit({ type: 'click' });
// Logger: { type: 'click' }
// Analytics: { type: 'click' }

Event Stream from DOM

function fromEvent(target, eventName) {
  return new Observable(observer => {
    const handler = (event) => observer.next(event);
    target.addEventListener(eventName, handler);

    return () => {
      target.removeEventListener(eventName, handler);
    };
  });
}

const button = document.getElementById('button');

fromEvent(button, 'click')
  .map(event => event.target.textContent)
  .subscribe({
    next: (text) => console.log('Button text:', text)
  });

BehaviorSubject

class BehaviorSubject {
  constructor(initialValue) {
    this.value = initialValue;
    this.observers = [];
  }

  getValue() {
    return this.value;
  }

  setValue(value) {
    this.value = value;
    this.emit(value);
  }

  subscribe(observer) {
    observer.next(this.value); // Emit current value immediately
    this.observers.push(observer);

    return () => {
      this.observers = this.observers.filter(o => o !== observer);
    };
  }

  emit(value) {
    this.observers.forEach(obs => obs.next(value));
  }
}

const count$ = new BehaviorSubject(0);

count$.subscribe({
  next: (value) => console.log('Count:', value)
});
// Immediately logs: Count: 0

count$.setValue(1);
// Logs: Count: 1

RxJS Example

// npm install rxjs

const { of, fromEvent, interval } = require('rxjs');
const { map, filter, debounceTime, switchMap } = require('rxjs/operators');

// From values
of(1, 2, 3)
  .pipe(
    map(x => x * 2),
    filter(x => x > 3)
  )
  .subscribe(console.log); // 4, 6

// From DOM event
const input = document.getElementById('search');

fromEvent(input, 'input')
  .pipe(
    debounceTime(500),
    map(event => event.target.value),
    switchMap(query => fetchResults(query))
  )
  .subscribe(results => {
    console.log('Search results:', results);
  });

// Interval stream
interval(1000)
  .pipe(
    map(i => i + 1),
    filter(i => i % 2 === 0)
  )
  .subscribe(console.log); // 2, 4, 6, ...

Custom Operator

function debounce(delay) {
  return function debounceOperator(source) {
    return new Observable(observer => {
      let timeoutId;

      return source.subscribe({
        next: (value) => {
          clearTimeout(timeoutId);
          timeoutId = setTimeout(
            () => observer.next(value),
            delay
          );
        },
        error: (err) => observer.error(err),
        complete: () => observer.complete()
      });
    });
  };
}

// Usage with custom operator
stream
  .pipe(debounce(500))
  .subscribe(console.log);

Best Practices

// 1. Unsubscribe to prevent memory leaks
const unsubscribe = stream.subscribe(observer);
// Later...
unsubscribe();

// 2. Use takeUntil to auto-unsubscribe
const stop$ = new Subject();
stream.pipe(takeUntil(stop$)).subscribe(observer);
stop$.next(); // Unsubscribe

// 3. Handle errors
stream.subscribe({
  next: (value) => console.log(value),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('Done')
});

// 4. Avoid nested subscriptions (use flatMap, switchMap, etc)
// ❌ Bad: Nested subscriptions
stream.subscribe(value => {
  otherStream.subscribe(otherValue => {
    console.log(value, otherValue);
  });
});

// ✅ Good: Use switchMap
stream.pipe(
  switchMap(value =>
    otherStream.pipe(
      map(otherValue => ({ value, otherValue }))
    )
  )
).subscribe(({ value, otherValue }) => {
  console.log(value, otherValue);
});