Multicasting with a Generic Subject

Hello! Welcome back to our course on building a modern JavaScript framework.

In our last lesson, we created several factory functions (of, from, interval) to produce "cold," "unicast" observables. This means each time you subscribe, you get a new, independent execution of that observable's logic. This is great for many scenarios, but what if you want multiple parts of your application to listen to the exact same stream of values? Or what if you need to programmatically push values into a stream, much like an event emitter?

Today, we'll address this by implementing a Subject. The learning outcome for this lesson is to implement a generic Subject class that acts as both an Observer and an Observable to multicast values. A Subject is a special hybrid that sits at the heart of many reactive patterns, enabling you to share data streams and bridge imperative code with your reactive world.

1. From Unicast to Multicast: The Need for a Subject

Our current Observable is unicast. Imagine our interval(1000) observable. If two different parts of your code subscribe to it, each subscription will start its own independent setInterval timer. They will emit values in parallel, but they won't be synchronized.

Multicasting, by contrast, means a single observable execution is shared among multiple subscribers. When the source emits a value, all subscribers receive it. This is the core idea behind the Observer design pattern.

To achieve this, we need a Subject.

Now, let's see a practical demonstration. The following video shows how a Subject is used in RxJS. The concepts are directly applicable to what we are building.

2. The Anatomy of a Subject

Based on the resources, we can define the structure of our Subject class. It's a fascinating hybrid that combines interfaces we've already built.

• It's an Observable: It must have a subscribe method so that consumers can listen to its values. We can achieve this by having our Subject class extend our Observable class.
• It's an Observer: It must have next(v), error(e), and complete() methods, allowing it to be on the receiving end of a stream. We'll implement our Observer interface.
• It's a Multicaster: Internally, it needs to maintain a list of all the observers that have subscribed to it. When next(), error(), or complete() is called on the subject, it will iterate through this list and call the corresponding method on each subscribed observer.

This UML diagram provides a clear visual of the relationships we are about to implement. Our Subject class will embody both the IObservable (by extending Observable) and IObserver roles.

3. Your Task: Implementing the Subject Class

Let's translate this design into code.

1. Create the file and class skeleton.
   In your src directory, create a new file named subject.ts. Inside, define the Subject class. It should extend Observable<T> and implement our Observer<T> interface. It also needs an internal array to hold its observers and a flag to track its state.

   // src/subject.ts
   import { Observable } from './observable';
   import { Observer } from './observer';
   import { Subscription } from './subscription';
   
   export class Subject<T> extends Observable<T> implements Observer<T> {
     private observers: Observer<T>[] = [];
     private isStopped = false;
   
     constructor() {
       // We call super with a producer that subscribes the subject to itself.
       // This allows `pipe` to work correctly.
       super(observer => {
         return this.subscribe(observer);
       });
     }
     
     // Methods to be implemented...
   }
   

   Note: The constructor here is a bit of a mind-bender. By passing this.subscribe to the parent Observable constructor, we ensure that when someone uses the pipe() method on a Subject, the resulting new Observable correctly subscribes back to the Subject itself.

2. Implement the Observer interface methods.
   These methods are the "input" side of the Subject. When they are called, the Subject multicasts the notification to all of its registered observers. We must also check the isStopped flag to ensure no notifications are sent after an error or completion.

   // Inside the Subject<T> class in src/subject.ts
   
   public next(value: T): void {
     if (this.isStopped) {
       return;
     }
     // We create a copy of the observers array in case
     // a subscriber unsubscribes during the next() call.
     for (const observer of [...this.observers]) {
       observer.next(value);
     }
   }
   
   public error(err: any): void {
     if (this.isStopped) {
       return;
     }
     this.isStopped = true;
     for (const observer of [...this.observers]) {
       observer.error(err);
     }
     // Clear observers after termination
     this.observers = [];
   }
   
   public complete(): void {
     if (this.isStopped) {
       return;
     }
     this.isStopped = true;
     for (const observer of [...this.observers]) {
       observer.complete();
     }
     // Clear observers after termination
     this.observers = [];
   }
   

3. Override the subscribe method.
   The subscribe method from the parent Observable class is not suitable because it expects a producer function. We need to override it to manage our internal list of observers. This method is the "output" side of the Subject.

   // Inside the Subject<T> class in src/subject.ts
   
   public subscribe(observer: Observer<T>): Subscription {
     if (this.isStopped) {
       // If the subject is already stopped, immediately notify
       // the new subscriber of the error or completion.
       // For simplicity, we'll just complete it. A more advanced
       // implementation might store the error.
       observer.complete();
       return new Subscription(() => {});
     }
   
     this.observers.push(observer);
   
     // Return a subscription that, when unsubscribed,
     // removes the observer from our list.
     const teardown = () => {
       const index = this.observers.indexOf(observer);
       if (index !== -1) {
         this.observers.splice(index, 1);
       }
     };
   
     return new Subscription(teardown);
   }
   

4. Your Task: Testing the Subject

With the implementation complete, let's write tests to verify its behavior. Create a new test file src/subject.test.ts.

// src/subject.test.ts
import { describe, it, expect, vi } from 'vitest';
import { Subject } from './subject';

describe('Subject', () => {
  it('should multicast values to multiple observers', () => {
    const subject = new Subject<number>();
    const observerA = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };
    const observerB = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    subject.subscribe(observerA);
    subject.subscribe(observerB);

    subject.next(1);
    subject.next(2);

    expect(observerA.next).toHaveBeenCalledTimes(2);
    expect(observerA.next).toHaveBeenCalledWith(1);
    expect(observerA.next).toHaveBeenCalledWith(2);

    expect(observerB.next).toHaveBeenCalledTimes(2);
    expect(observerB.next).toHaveBeenCalledWith(1);
    expect(observerB.next).toHaveBeenCalledWith(2);
  });

  it('should stop multicasting after complete is called', () => {
    const subject = new Subject<number>();
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    subject.subscribe(observer);

    subject.next(1);
    subject.complete();
    subject.next(2); // This should be ignored

    expect(observer.next).toHaveBeenCalledTimes(1);
    expect(observer.next).toHaveBeenCalledWith(1);
    expect(observer.complete).toHaveBeenCalledTimes(1);
  });

  it('should stop multicasting after error is called', () => {
    const subject = new Subject<number>();
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };
    const testError = new Error('test');

    subject.subscribe(observer);

    subject.next(1);
    subject.error(testError);
    subject.next(2); // This should be ignored

    expect(observer.next).toHaveBeenCalledTimes(1);
    expect(observer.error).toHaveBeenCalledTimes(1);
    expect(observer.error).toHaveBeenCalledWith(testError);
    expect(observer.complete).not.toHaveBeenCalled();
  });

  it('should stop sending values to an observer after it unsubscribes', () => {
    const subject = new Subject<number>();
    const observerA = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };
    const observerB = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    const subA = subject.subscribe(observerA);
    subject.subscribe(observerB);

    subject.next(1);
    subA.unsubscribe();
    subject.next(2);

    expect(observerA.next).toHaveBeenCalledTimes(1);
    expect(observerA.next).toHaveBeenCalledWith(1);

    expect(observerB.next).toHaveBeenCalledTimes(2);
    expect(observerB.next).toHaveBeenCalledWith(1);
    expect(observerB.next).toHaveBeenCalledWith(2);
  });
});

Run your tests to confirm that your Subject behaves as expected.

Conclusion

Congratulations! You have successfully implemented a Subject, a cornerstone of reactive programming. You've created a class that is both an Observable and an Observer, enabling it to multicast values to many subscribers.

Key Takeaways:

• Subject as a Hybrid: A Subject uniquely combines the Observable and Observer interfaces.
• Multicasting: Unlike unicast observables, a Subject shares a single execution among all its subscribers.
• Imperative Bridge: Subjects allow you to imperatively call .next(), .error(), and .complete(), acting as a bridge between traditional event-driven code and the reactive world.
• State Management: A Subject must manage its internal state, keeping a list of observers and tracking whether it has been stopped to ensure correct behavior.

Preview of the Next Lesson:
Now that we have a Subject, we can unlock one of its most powerful use cases: sharing a single subscription to a cold observable. In the next lesson, we will implement a share operator. This operator will use a Subject internally to convert any cold, unicast observable into a hot, multicast one, preventing redundant work and synchronizing data for all consumers.