Creating Observables from Scratch

Hello! Welcome back.

In our previous lessons, we explored RxJS's built-in creation operators, learning to generate streams from static data (of, from), events (fromEvent), and time-based sequences (interval, timer). We also saw how defer gives us fine-grained control over when an Observable's logic executes. These operators cover the vast majority of common use cases.

Today, we go to the very heart of RxJS. We'll answer the question: "What if none of the built-in operators fit my needs?" This lesson is the capstone of our foundational module, where you will learn to construct an Observable from first principles. This is a key skill for advanced, real-world development, enabling you to wrap any asynchronous source—be it a third-party library's callbacks, a WebSocket, or a custom event system—into a well-behaved, composable RxJS stream.

Our learning outcome is to implement a custom Observable using the Observable constructor to wrap a non-standard asynchronous source.

1. The Anatomy of an Observable

At its core, an Observable is a blueprint for a data stream. The new Observable() constructor is the tool we use to define this blueprint. It takes a single argument: a function, often called the subscribe function.

This function is the "producer" logic. It's where you define what happens when a consumer subscribes. RxJS invokes this function for each new subscription, passing in an observer object. This observer is the communication channel to the consumer, equipped with three methods:

• observer.next(value): Pushes a new value to the consumer.
• observer.error(err): Notifies the consumer that an error has occurred and terminates the stream.
• observer.complete(): Notifies the consumer that the stream has finished successfully.

After error or complete is called, the stream is considered finished, and no further next calls will be delivered. This is known as the Observable contract.

To start, let's get the official definition from the RxJS documentation.

As you read, you can see that an Observable is essentially a function that describes how to produce values for an observer. It's a lazy computation that only runs when subscribe() is called.

Here is a simple synchronous example to illustrate the mechanics:

import { Observable } from 'rxjs';

// 1. Define the blueprint
const myObservable = new Observable(subscriber => {
  console.log('Observable logic executed!');
  subscriber.next(1);
  subscriber.next(2);
  // This would throw an error and be caught by the error path
  // throw new Error('Something went wrong!'); 
  subscriber.next(3);
  subscriber.complete();
  subscriber.next(4); // This will not be delivered
});

// 2. Subscribe to trigger the execution
console.log('Before subscribe');
myObservable.subscribe({
  next: value => console.log('Received value:', value),
  error: err => console.error('Caught error:', err.message),
  complete: () => console.log('Stream completed.')
});
console.log('After subscribe');

2. Building a Custom Observable: A Guided Example

Now that we have the basic theory, let's see it in a more practical, asynchronous context. The following video provides an excellent step-by-step guide to building a custom Observable.

A quick note on syntax: The video uses Observable.create(), which was the syntax in older versions of RxJS. As of RxJS 6, this has been replaced by the new Observable() constructor we are using. The underlying concept is identical, so the video's explanation remains perfectly valid.

This walkthrough solidifies the idea that the subscribe function you provide to the constructor is a container for any logic—synchronous or asynchronous—that produces values.

3. The Teardown Logic: Preventing Resource Leaks

The video you just watched briefly mentioned unsubscribing. This is a critical aspect of creating custom Observables, especially for long-lived or infinite streams (like interval, WebSockets, or DOM events).

What happens when a consumer is no longer interested in the stream and unsubscribes? If your producer logic set up a timer (setInterval) or opened a connection (like a WebSocket), that resource will continue to run in the background, consuming memory and CPU. This is a memory leak.

To solve this, the Observable constructor has a powerful feature: your subscribe function can return a teardown function. This function contains the cleanup logic. RxJS guarantees that this function will be called automatically when:

1. The consumer calls unsubscribe().
2. The stream terminates via complete() or error().

Let's return to the official documentation for the canonical explanation of this mechanism.

This teardown mechanism is the key to creating robust, well-behaved Observables that safely manage the lifecycle of any underlying resource.

4. Practical Application: Wrapping Non-Standard APIs

Now, let's put everything together. The primary reason to create a custom Observable is to act as an adapter, bridging the gap between the reactive world of RxJS and other asynchronous patterns like callbacks, event emitters, or browser APIs.

The following resource provides excellent, real-world examples of this pattern.

These examples are perfect illustrations of the learning outcome. You are taking a "non-standard" source (one that doesn't speak Observable) and wrapping it in a way that it becomes a fully-featured, safe, and reusable RxJS stream.

Let's try a small exercise. Imagine you have a legacy function that uses a classic callback pattern:

/**
 * Simulates a legacy async function that takes time to process data.
 * @param {string} data The input data.
 * @param {(result: string) => void} onSuccess Callback for success.
 * @param {(error: string) => void} onError Callback for failure.
 */
function legacyAsyncOperation(data, onSuccess, onError) {
  const processingTime = 1000 + Math.random() * 1000;
  setTimeout(() => {
    if (Math.random() > 0.2) {
      onSuccess(`Processed: ${data}`);
    } else {
      onError('Failure during processing');
    }
  }, processingTime);
}

How would you write a function fromLegacy(data) that wraps this operation and returns an Observable?

import { Observable } from 'rxjs';

function fromLegacy(data) {
  return new Observable(observer => {
    // We don't need to track any resource ID here, so no teardown is needed.
    // The setTimeout is self-contained and will be garbage collected.
    // If this were a persistent connection, we would need teardown logic.

    legacyAsyncOperation(
      data,
      (result) => {
        observer.next(result);
        observer.complete(); // The operation is a one-shot deal.
      },
      (error) => {
        observer.error(error); // The operation failed.
      }
    );

    // Note: No teardown function is returned because setTimeout doesn't need manual clearing
    // once it has fired. If the subscriber unsubscribes *before* it fires, the timeout
    // will still run, but the observer methods will do nothing, preventing side effects.
    // For a more robust implementation, you could clear the timeout in the teardown.
    
    /* A more robust version with teardown:
    return new Observable(observer => {
        const timeoutId = setTimeout(() => { ... }, processingTime);

        return () => {
            clearTimeout(timeoutId);
        }
    });
    */
  });
}

// Usage:
fromLegacy('my-data-123').subscribe({
  next: console.log,
  error: console.error,
  complete: () => console.log('Legacy operation finished.')
});

Conclusion

This lesson completes our journey through the foundations of creating Observables. While you will most often use the high-level creation operators we learned previously, understanding how to build one from scratch is an invaluable skill for an advanced RxJS practitioner. It gives you the power to integrate RxJS into any project, regardless of its existing asynchronous architecture.

Key Takeaways:

• You can create a custom Observable with new Observable(subscribeFn).
• The subscribeFn is your producer logic. It receives an observer object to communicate with the consumer via next(), error(), and complete().
• This function is executed lazily for each new subscription.
• To prevent memory leaks and manage resources, you can return a teardown function from your subscribeFn. This function is called on unsubscription, completion, or error.
• This pattern is the fundamental way to adapt any non-reactive, asynchronous API into a well-behaved RxJS Observable.

In our next module, "Core Operators and Stream Transformation," we will shift our focus from creating streams to manipulating them. You'll learn how to transform, filter, combine, and manage streams of data, which is where the true expressive power of RxJS begins to shine. We'll start by learning how to inspect and transform values with tap and map.