Reactive Programming Guide

Production-grade guide to reactive programming guide covering architecture patterns, implementation strategies, testing approaches, and operational best practices for enterprise engineering teams.

This is a reference for implementing reactive programming patterns in systems that process streams of data or events. It covers how to design and implement systems that respond to changes in real-time, manage backpressure, and handle asynchronous data flows without blocking threads.

Creating Reactive Data Streams

Transforming Data in Real-Time

To create a stream processor that filters and transforms incoming events:

const { Observable } = require('rxjs');

const eventStream = new Observable(subscriber => {
  // Simulate receiving events from an external source
  setInterval(() => {
    const event = {
      id: Math.random(),
      timestamp: Date.now(),
      data: Math.random() > 0.5 ? 'success' : 'failure'
    };
    subscriber.next(event);
  }, 100);
});

const filteredStream = eventStream
  .filter(event => event.data === 'success')
  .map(event => ({
    ...event,
    processedAt: Date.now()
  }));

filteredStream.subscribe({
  next: (event) => console.log('Processed:', event),
  error: (err) => console.error('Stream error:', err)
});

The key configuration is the subscription's next handler. If you don't provide an error handler, unhandled errors in the stream will silently terminate the subscription.

Managing Backpressure

When a downstream consumer cannot keep up with data flow, implement backpressure handling:

const { Observable } = require('rxjs');

const slowConsumerStream = new Observable(subscriber => {
  const interval = setInterval(() => {
    subscriber.next({ data: 'chunk', timestamp: Date.now() });
  }, 10);

  return () => clearInterval(interval);
});

// Throttle to prevent overwhelming the consumer
slowConsumerStream
  .pipe(
    throttleTime(500),
    bufferTime(1000)
  )
  .subscribe(chunk => {
    console.log('Processing batch:', chunk.length, 'items');
  });

The throttleTime operator prevents emitting more than one value per 500ms. The bufferTime collects values into arrays over a time window. Both operators must be applied in the correct order for expected behavior.

Handling Asynchronous Operations

Chaining Async Operations

To chain operations where each depends on the previous result:

const { Observable } = require('rxjs');

const asyncStream = new Observable(subscriber => {
  const data = ['a', 'b', 'c'];
  data.forEach(item => subscriber.next(item));
});

asyncStream
  .mergeMap(item => 
    fetch(`/api/data/${item}`)
      .then(response => response.json())
      .catch(err => {
        console.error('Fetch error:', err);
        return { error: true, item };
      })
  )
  .subscribe(result => {
    if (result.error) {
      console.error('Processing failed for item:', result.item);
    } else {
      console.log('Success:', result);
    }
  });

The mergeMap operator is essential here. Using flatMap instead will produce the same result but with less explicit control over the inner observable's subscription lifecycle.

Error Propagation in Async Chains

When an operation fails in a chain, it can be handled locally or let bubble up:

const { Observable } = require('rxjs');

const failingStream = new Observable(subscriber => {
  subscriber.next({ id: 1 });
  subscriber.next({ id: 2 });
  subscriber.error(new Error('Critical failure'));
});

failingStream
  .catchError(error => {
    console.error('Caught error:', error.message);
    return Observable.of({ id: 'fallback', error: true });
  })
  .subscribe({
    next: data => console.log('Received:', data),
    error: err => console.error('Uncaught error:', err)
  });

The catchError operator must return an observable. If you return a plain value, it will be treated as the first value in a new stream but not properly subscribed to.

Resource Management

Proper Cleanup of Observables

To prevent memory leaks when creating long-running streams:

const { Observable } = require('rxjs');

const periodicStream = new Observable(subscriber => {
  const timer = setInterval(() => {
    subscriber.next(Date.now());
  }, 1000);

  // Return cleanup function
  return () => {
    clearInterval(timer);
    console.log('Stream cleanup completed');
  };
});

const subscription = periodicStream.subscribe(value => {
  console.log('Value:', value);
});

// Later, when you want to stop:
subscription.unsubscribe();

The cleanup function in the observable factory is crucial. Without it, timers and event listeners will accumulate and cause memory leaks.

Managing Multiple Subscriptions

When you need to coordinate multiple streams:

const { Observable } = require('rxjs');

const stream1 = Observable.interval(1000);
const stream2 = Observable.interval(500);

Observable.combineLatest([stream1, stream2])
  .subscribe(([val1, val2]) => {
    console.log('Combined values:', val1, val2);
  });

The combineLatest operator waits for at least one emission from each source before emitting. If you want to emit as soon as any source emits, use merge instead.

Common Pitfalls and Failure Modes

Silent Subscription Failures

If an observable's subscription doesn't handle errors properly:

const { Observable } = require('rxjs');

const problematicStream = new Observable(subscriber => {
  subscriber.next('data');
  subscriber.error(new Error('Something went wrong'));
});

// Missing error handler - silent failure
problematicStream.subscribe(value => {
  console.log('Value:', value);
});

This code will silently terminate the subscription when the error occurs, and no error will be logged. The subscribe method accepts three arguments: next, error, and complete. All are optional but should be explicitly provided for production code.

Memory Leaks from Unsubscribed Streams

When streams are created but not properly managed:

const { Observable } = require('rxjs');

function createInfiniteStream() {
  return new Observable(subscriber => {
    const timer = setInterval(() => {
      subscriber.next(Date.now());
    }, 1000);
    
    // No cleanup function returned
  });
}

// This creates a memory leak:
createInfiniteStream().subscribe(value => {
  console.log(value);
});

The stream will keep running even after the subscription is no longer needed. Always return a cleanup function from the observable factory.

Incorrect Operator Order

The order of operators affects the result:

const { Observable } = require('rxjs');

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

// Wrong order - filters first, then maps
source
  .filter(x => x > 2)
  .map(x => x * 2)
  .subscribe(result => console.log(result));
// Output: 6, 8, 10

// Correct order - maps first, then filters
source
  .map(x => x * 2)
  .filter(x => x > 4)
  .subscribe(result => console.log(result));
// Output: 6, 8, 10

The difference is subtle but important. Mapping before filtering reduces the number of operations while keeping the same final result.

Race Conditions in Stream Processing

When multiple operations can interfere with each other:

const { Observable } = require('rxjs');

const raceStream = new Observable(subscriber => {
  setTimeout(() => subscriber.next('slow'), 1000);
  setTimeout(() => subscriber.next('fast'), 500);
});

raceStream
  .first() // Takes only the first emission
  .subscribe(value => console.log('First value:', value));
// Output: First value: fast

The first() operator is crucial here. Without it, you might get multiple emissions or miss important data.

Testing Reactive Systems

Testing Stream Behavior

To verify that streams behave correctly:

const { Observable } = require('rxjs');
const { TestScheduler } = require('rxjs/testing');

const scheduler = new TestScheduler((actual, expected) => {
  // Custom assertion
  expect(actual).toEqual(expected);
});

scheduler.run(({ expectObservable }) => {
  const source = Observable.of(1, 2, 3);
  const result = source.map(x => x * 2);
  
  expectObservable(result).toBe('(abc|)', {
    a: 2,
    b: 4,
    c: 6
  });
});

The test scheduler is essential for deterministic testing. Using setTimeout or setInterval directly in tests will make them flaky.

Testing Error Handling

To verify error scenarios:

const { Observable } = require('rxjs');

const errorStream = new Observable(subscriber => {
  subscriber.next('data');
  subscriber.error(new Error('Test error'));
});

errorStream
  .catchError(err => Observable.of('fallback'))
  .subscribe({
    next: value => console.log('Value:', value),
    error: err => console.error('Error:', err.message)
  });

The catchError operator must be tested for both success and error paths. Without it, errors can silently break the entire stream.

This guide covers essential patterns for building reactive systems. The key is understanding that streams are stateful and require careful management of subscriptions, resource cleanup, and error handling.

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.