Implementing `subscribe` for Observable-Observer Connection

Hello! Welcome back.

In our last lesson, we defined the core structure for our Observable class, establishing that its behavior is defined by a producer function passed to its constructor. We now have all our foundational pieces: the Observer interface (the consumer), the Subscription class (the lifecycle manager), and the Observable class (the source).

Today, we will implement the subscribe method. This is the engine that brings all three components together, activating the data stream and enabling resource management.

This lesson directly addresses the learning outcome: Implement the subscribe method to manage the connection between an Observer and the Observable's producer function, returning a Subscription.

1. Recap and The Role of subscribe

Let's quickly recap:

• Observable: Holds a "recipe" (the producer function) for creating notifications. It's "cold," meaning the recipe isn't executed until someone subscribes.
• Observer: An object with next, error, and complete methods, ready to receive notifications.
• Subscription: An object that represents an ongoing execution and contains teardown logic for cleanup.

The subscribe method is the "ignition key." When a consumer calls observable.subscribe(observer), it's instructing the observable to execute its producer function and start sending notifications to that specific observer. The method's other critical job is to return a Subscription so the consumer can later call unsubscribe() to stop the flow of data and clean up any resources.

To see this relationship in action within RxJS, the library that inspires our work, let's watch a brief video segment.

2. The Need for a "Safe" Subscriber

Before we jump into the implementation, let's consider a potential problem. What if a producer function is poorly written?

const badProducer = (observer) => {
  observer.next(1);
  observer.complete();
  observer.next(2); // This violates the Observable contract!
};

The Observable contract states that no next or error notifications should be delivered after complete has been called. Likewise, nothing should be delivered after an error. A raw Observer passed directly to a producer has no protection against such behavior.

To build a robust library, we need to guarantee this contract. The solution is to wrap the user-provided Observer in a protective layer that enforces these rules. We'll call this a SafeSubscriber.

Ben Lesh, the lead of the RxJS team, wrote an excellent article that walks through building an observable from scratch. It explains the motivation for this safety layer and how to implement it.

This SafeSubscriber will act as a proxy. Our subscribe method will create it, and the producer function will talk to the SafeSubscriber, which will then safely forward notifications to the original Observer.

3. Implementing the subscribe Method and SafeSubscriber

Now we have all the concepts needed to build a complete, robust subscribe implementation. The process will be:

1. The subscribe method is called with an observer.
2. It creates a new Subscription object.
3. It creates a SafeSubscriber that wraps the user's observer and is linked to the new subscription.
4. It executes the Observable's private _producer function, passing the SafeSubscriber as the argument.
5. The producer may return a teardown function. If it does, this function is added to the subscription.
6. Finally, the subscribe method returns the subscription to the caller.

The article from Ben Lesh provides a complete, integrated implementation of this logic.

4. Your Task: Finalize the Core Implementation

Let's translate this design into our project. We'll create the SafeSubscriber and then implement the subscribe method in our Observable class.

1. Create src/SafeSubscriber.ts

Create a new file for our SafeSubscriber. This class is an internal detail of our library, but crucial for its correctness.

// src/SafeSubscriber.ts

import { Observer } from "./Observer";
import { Subscription } from "./Subscription";

export class SafeSubscriber<T> implements Observer<T> {
  private isClosed = false;

  constructor(
    private destination: Observer<T>,
    private subscription: Subscription
  ) {
    // When the subscription is unsubscribed, this subscriber is closed.
    subscription.add(() => (this.isClosed = true));
  }

  next(value: T): void {
    if (!this.isClosed) {
      // We use a try/catch here to ensure that if the user's `next` handler
      // throws an error, we can catch it and handle it gracefully.
      try {
        this.destination.next(value);
      } catch (err) {
        this.error(err);
      }
    }
  }

  error(err: unknown): void {
    if (!this.isClosed) {
      this.isClosed = true;
      this.destination.error(err);
      // After an error, the subscription is considered finished.
      this.subscription.unsubscribe();
    }
  }

  complete(): void {
    if (!this.isClosed) {
      this.isClosed = true;
      this.destination.complete();
      // After completion, the subscription is considered finished.
      this.subscription.unsubscribe();
    }
  }
}

Note: We've made a slight improvement over the article's version by wrapping the destination.next() call in a try...catch block. This is a standard practice in robust reactive libraries to ensure that an error in a consumer's next handler properly terminates the stream.

2. Update src/Observable.ts

Now, import the SafeSubscriber and implement the subscribe method.

// src/Observable.ts

import { Observer } from "./Observer";
import { Subscription } from "./Subscription";
import { SafeSubscriber } from "./SafeSubscriber"; // <-- Import SafeSubscriber

export type Producer<T> = (
  observer: Observer<T>
) => (() => void) | void;

export class Observable<T> {
  private _producer: Producer<T>;

  constructor(producer: Producer<T>) {
    this._producer = producer;
  }

  /**
   * Subscribes to the stream, activating the producer function and connecting
   * it to the provided observer.
   * @param observer The object that will receive notifications.
   * @returns A Subscription object that can be used to unsubscribe.
   */
  subscribe(observer: Observer<T>): Subscription {
    // 1. Create the Subscription that will be returned to the consumer.
    const subscription = new Subscription();
    
    // 2. Create the SafeSubscriber to wrap the user's observer.
    const subscriber = new SafeSubscriber(observer, subscription);

    // 3. Execute the producer, passing the safe subscriber.
    //    Capture the returned teardown logic.
    const teardown = this._producer(subscriber);

    // 4. Add the producer's teardown logic to the subscription.
    if (teardown) {
      subscription.add(teardown);
    }

    // 5. Return the subscription to the consumer.
    return subscription;
  }
}

With this implementation, our core observable mechanism is complete. We have a system where calling subscribe safely executes the producer, delivers values, and provides a reliable way to tear down the stream.

To make the concept of "teardown logic" more concrete, the learnrxjs.io website has excellent examples of wrapping real-world asynchronous APIs that require cleanup. For instance, when you create an observable from setInterval, the teardown function must call clearInterval.

Conclusion

In this lesson, we implemented the heart of our reactive library: the Observable.subscribe method. We connected the Observable, Observer, and Subscription components into a functioning, robust system.

Key Takeaways:

• The subscribe method activates an observable's producer function.
• A SafeSubscriber is a crucial internal component that wraps the user's observer to enforce the observable contract (e.g., no notifications after completion/error).
• The subscribe method orchestrates the entire process: creating a Subscription, wrapping the Observer in a SafeSubscriber, executing the producer, and managing the teardown logic.
• The method returns a Subscription, giving the consumer full control over the lifecycle of the stream execution.

Preview of the Next Lesson:

Our core observable is now functionally complete. But how do we know it works correctly in all scenarios (synchronous, asynchronous, errors, etc.)? In the next lesson, we will use Vitest to write a comprehensive suite of unit tests to validate the entire observable lifecycle, ensuring our foundation is solid before we start building operators on top of it.