Building Custom Reactive Stream Libraries in JavaScript: Core Techniques and Best Practices

#programming #devto #javascript #softwareengineering

As a best-selling author, I invite you to explore my books on Amazon. Don't forget to follow me on Medium and show your support. Thank you! Your support means the world!

Building reactive stream libraries in JavaScript requires thoughtful approaches to asynchronous data flow. I've found that focusing on core techniques creates resilient systems capable of handling complex scenarios. Let me share practical methods that have proven effective in my work.

Observable creation starts with flexible data source handling. I implement factories for common scenarios like DOM events or timed intervals. The key is lazy initialization - nothing activates until subscription occurs. This approach conserves resources until needed. Promise conversions require special attention to single-value emission.

class Observable {
  static fromPromise(promise) {
    return new Observable(observer => {
      promise.then(
        value => {
          observer.next(value);
          observer.complete();
        },
        error => observer.error(error)
      );
      return () => {};
    });
  }
}

// Creating from fetch API
const apiStream = Observable.fromPromise(
  fetch('https://api.example.com/data')
).subscribe({
  next: response => console.log('Data received'),
  error: err => console.error('Fetch failed', err)
});

Operator composition transforms streams through chainable functions. I build operators like map and filter that process values without altering source streams. Pipelining creates readable transformation sequences. Custom operators encapsulate domain logic while maintaining error propagation. Each operator returns a new observable, preserving the original stream's integrity.

function take(count) {
  return source => new Observable(observer => {
    let taken = 0;
    const sub = source.subscribe({
      next: val => {
        if (taken++ < count) {
          observer.next(val);
          if (taken === count) {
            observer.complete();
            sub.unsubscribe();
          }
        }
      },
      error: err => observer.error(err),
      complete: () => observer.complete()
    });
    return sub;
  });
}

// Usage
Observable.interval(500)
  .pipe(take(5))
  .subscribe(val => console.log(val)); // Logs 0-4

Subscription management handles resource cleanup through disposables. I design subscription trees that cascade cancellations through operator chains. Finalization hooks release resources whether streams complete or error. This prevents memory leaks from forgotten event listeners or timers.

class Subscription {
  constructor() {
    this.teardowns = new Set();
  }

  add(teardown) {
    this.teardowns.add(teardown);
  }

  unsubscribe() {
    this.teardowns.forEach(teardown => teardown());
    this.teardowns.clear();
  }
}

// In observable
const obs = new Observable(observer => {
  const timer = setInterval(() => observer.next('tick'), 1000);
  return () => clearInterval(timer);
});

const sub = obs.subscribe(console.log);
setTimeout(() => sub.unsubscribe(), 5000); // Cleans up timer

Multicasting efficiently shares streams among multiple observers. Subjects broadcast values while replay capabilities cache emissions for late subscribers. Reference-counted connections automatically manage shared subscriptions. This prevents duplicate work when multiple components consume the same stream.

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

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

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

// Usage
const sensorData$ = new Subject();
sensorData$.subscribe(val => console.log('Display 1:', val));
sensorData$.subscribe(val => console.log('Display 2:', val));

sensorData$.next(42); // Both displays log 42

Backpressure management balances producer and consumer speeds. I implement buffering for lossless handling with slower consumers. Sampling provides lossy approaches for high-frequency streams. Pause/resume mechanisms coordinate flow control through explicit signaling.

function bufferTime(windowTime) {
  return source => new Observable(observer => {
    let buffer = [];
    const timer = setInterval(() => {
      if (buffer.length) {
        observer.next(buffer);
        buffer = [];
      }
    }, windowTime);

    return source.subscribe({
      next: val => buffer.push(val),
      error: err => observer.error(err),
      complete: () => {
        clearInterval(timer);
        observer.complete();
      }
    });
  });
}

// Throttle mouse events
Observable.fromEvent(document, 'mousemove')
  .pipe(bufferTime(100))
  .subscribe(positions => console.log(positions));

Scheduler integration abstracts time management. Virtual time schedulers enable deterministic testing. Animation frame schedulers synchronize with browser repaints. Queue schedulers manage microtask execution while immediate schedulers handle synchronous operations.

class Scheduler {
  constructor(scheduleFn) {
    this.schedule = scheduleFn;
  }
}

const animationFrameScheduler = new Scheduler(callback => {
  const frameId = requestAnimationFrame(callback);
  return () => cancelAnimationFrame(frameId);
});

// Smooth animation stream
Observable.interval(0, animationFrameScheduler)
  .pipe(map(() => performance.now()))
  .subscribe(timestamp => animateElement(timestamp));

Error recovery maintains stream stability during failures. Catch operators switch to backup streams while retry strategies attempt recovery with configurable delays. Finally operators execute cleanup regardless of outcome. These mechanisms prevent total pipeline failures from single errors.

function retry(maxAttempts, delay) {
  return source => new Observable(observer => {
    let attempts = 0;
    function subscribe() {
      source.subscribe({
        next: val => observer.next(val),
        error: err => {
          attempts++;
          if (attempts >= maxAttempts) {
            observer.error(err);
          } else {
            setTimeout(subscribe, delay);
          }
        },
        complete: () => observer.complete()
      });
    }
    subscribe();
  });
}

// Network request with retries
fetchDataStream.pipe(
  retry(3, 1000)
).subscribe(handleData);

Hot and cold streams exhibit different subscription behaviors. Cold streams restart for each subscriber while hot streams share live emissions. Publish/connect patterns control hot stream activation. Understanding these semantics ensures proper resource sharing and data freshness.

function publish(source) {
  const subject = new Subject();
  source.subscribe(subject);
  return subject;
}

// Turning cold stream hot
const cold$ = Observable.interval(1000);
const hot$ = publish(cold$);
hot$.connect(); // Starts emissions

setTimeout(() => {
  hot$.subscribe(console.log); // Shares existing interval
}, 3000);

Testing utilities verify stream behavior through visualization tools. Marble diagrams express time-based operations clearly. Virtual clocks enable deterministic time manipulation during tests. Assertion libraries validate complex emission sequences and timing constraints.

// Simplified marble testing
function testScheduler(assertion) {
  const events = [];
  const scheduler = new Scheduler(callback => {
    events.push(callback);
  });

  assertion(scheduler);

  while (events.length) {
    events.shift()();
  }
}

// Testing interval operator
testScheduler(scheduler => {
  const results = [];
  Observable.interval(10, scheduler)
    .pipe(take(3))
    .subscribe(val => results.push(val));

  scheduler.schedule(() => {
    assert.deepEqual(results, [0,1,2]);
  }, 30);
});

These techniques form a comprehensive approach to reactive stream implementation. Combining them creates systems that coordinate complex asynchronous operations while maintaining responsiveness. The patterns scale elegantly from simple event handlers to sophisticated data processing pipelines.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

101 Books is an AI-driven publishing company co-founded by author Aarav Joshi. By leveraging advanced AI technology, we keep our publishing costs incredibly low—some books are priced as low as $4—making quality knowledge accessible to everyone.

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

Create a feature flag in your IDE in 5 minutes with LaunchDarkly’s MCP server ⏰

How to create, evaluate, and modify flags from within your IDE or AI client using natural language with LaunchDarkly's new MCP server. Follow along with this tutorial for step by step instructions.

Read full post