Implementing a `share` Operator with `Subject`

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

In our last lesson, we implemented the Subject class, a powerful hybrid that acts as both an Observer and an Observable. This gave us the ability to multicast values—that is, to push a single value to multiple listeners simultaneously.

Today, we'll leverage the Subject to solve a fundamental problem in reactive programming: managing expensive or redundant operations. We will implement one of the most important operators in any reactive library: share.

The learning outcome for this lesson is to implement and test a share operator that uses a Subject to convert a cold, unicast observable into a hot, multicast observable.

1. The Problem: Cold, Unicast Observables

So far, all the observables we've created (of, from, interval) are cold and unicast.

• Cold: The observable's producer function (the logic that generates values) is executed for every new subscriber.
• Unicast: Each subscriber gets its own independent execution.

Imagine an observable that fetches data from a network API.

const apiData$ = new Observable(observer => {
  console.log('Fetching data from API...');
  fetch('https://api.example.com/data')
    .then(res => res.json())
    .then(data => {
      observer.next(data);
      observer.complete();
    })
    .catch(err => observer.error(err));
});

If two different components in your application subscribe to apiData$, the fetch call will be made twice. This is inefficient and can lead to inconsistent state if the API returns slightly different data each time.

The solution is to make the observable hot and multicast.

• Hot: The observable has a single, shared execution that starts independently of its subscribers (or when the first subscriber arrives).
• Multicast: All subscribers listen to that single, shared execution.

This is precisely what the share operator is for.

2. The Mechanics of share: Subject + Reference Counting

The share operator might seem magical, but it's built on two concepts, one of which we just implemented.

1. A Subject for Multicasting: The share operator subscribes to the original (cold) source observable just once. It then uses an internal Subject to broadcast the values from that single subscription to all of its own subscribers.

2. Reference Counting (refCount): To be efficient, the operator should only hold its subscription to the source as long as it has active listeners. It achieves this by counting its subscribers:

   • When the number of subscribers goes from 0 to 1, it subscribes to the source observable.
   • When the number of subscribers goes from 1 to 0 (i.e., the last subscriber unsubscribes), it unsubscribes from the source observable to clean up resources.

This combination makes the share operator both powerful and efficient.

For a quick visual demonstration of converting a cold observable to a hot one using share, the following video is very effective.

3. Your Task: Implementing the share Operator

Now, let's build it. A pipeable operator in our library is a function that returns another function. The inner function takes a source observable and returns a new Observable.

1. Create the file and function structure.
   In your src directory, create a new folder operators. Inside it, create a file named share.ts.

   // src/operators/share.ts
   import { Observable } from '../observable';
   import { Observer } from '../observer';
   import { Subject } from '../subject';
   import { Subscription } from '../subscription';
   
   export function share<T>(): (source: Observable<T>) => Observable<T> {
     // We need to keep track of the state across multiple subscriptions.
     // These variables are declared here to form a closure.
     let refCount = 0;
     let subject: Subject<T> | null = null;
     let sourceSubscription: Subscription | null = null;
   
     // The operator returns a function that takes the source observable.
     return (source: Observable<T>): Observable<T> => {
       // This function, in turn, returns a NEW observable.
       return new Observable<T>(observer => {
         // Implementation logic goes here...
       });
     };
   }
   

2. Implement the subscription logic.
   This is where the reference counting happens. When a new observer subscribes to the observable returned by share, we need to manage our shared Subject and the subscription to the source.

   Fill in the producer function for the new Observable:

   // Inside the `return new Observable<T>(observer => { ... });` block
   
   refCount++;
   
   // If this is the first subscriber, we need to set up the
   // shared subject and subscribe it to the source.
   if (!subject) {
     subject = new Subject<T>();
     sourceSubscription = source.subscribe(subject);
   }
   
   // Every subscriber, including the first, subscribes to the shared subject.
   const innerSub = subject.subscribe(observer);
   
   // The teardown logic is crucial for reference counting.
   // We return a subscription that cleans up when this specific observer unsubscribes.
   return new Subscription(() => {
     refCount--;
     innerSub.unsubscribe(); // Unlink the observer from the subject.
   
     // If this was the last subscriber, tear down the shared subscription.
     if (refCount === 0) {
       sourceSubscription?.unsubscribe();
       sourceSubscription = null;
       subject = null; // Reset for the next full subscription cycle.
     }
   });
   

   Your complete share.ts file should now contain the full implementation. The closure created by the share function ensures that refCount, subject, and sourceSubscription are shared among all subscribers to the same piped observable instance.

4. Your Task: Testing the share Operator

How can we be sure our operator works? We need to test two key behaviors: multicasting and reference counting.

Create a new test file src/operators/share.test.ts.

// src/operators/share.test.ts
import { describe, it, expect, vi } from 'vitest';
import { Observable } from '../observable';
import { share } from './share';

describe('share', () => {
  it('should multicast a cold observable, subscribing only once', () => {
    const producer = vi.fn(observer => {
      observer.next(1);
      observer.complete();
    });

    const source$ = new Observable(producer);
    const shared$ = source$.pipe(share());

    const observerA = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };
    const observerB = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    shared$.subscribe(observerA);
    shared$.subscribe(observerB);

    // The producer function should have been called only once.
    expect(producer).toHaveBeenCalledTimes(1);

    // Both observers should have received the same value.
    expect(observerA.next).toHaveBeenCalledWith(1);
    expect(observerB.next).toHaveBeenCalledWith(1);
    expect(observerA.complete).toHaveBeenCalled();
    expect(observerB.complete).toHaveBeenCalled();
  });

  it('should re-subscribe to the source if all subscribers unsubscribe and a new one subscribes', () => {
    const producer = vi.fn(observer => {
      observer.next(Math.random());
      observer.complete();
    });

    const source$ = new Observable(producer);
    const shared$ = source$.pipe(share());

    const observerA = { next: vi.fn(), complete: vi.fn() };
    const observerB = { next: vi.fn(), complete: vi.fn() };

    // First subscription cycle
    const subA = shared$.subscribe(observerA);
    subA.unsubscribe();

    // At this point, refCount should be 0, and the source subscription torn down.
    // A new subscription should trigger a new execution.
    const subB = shared$.subscribe(observerB);

    // The producer should have been called twice, once for each cycle.
    expect(producer).toHaveBeenCalledTimes(2);

    // The values should be different, proving new executions.
    expect(observerA.next).toHaveBeenCalledTimes(1);
    expect(observerB.next).toHaveBeenCalledTimes(1);
    expect(observerA.next.mock.calls[0][0]).not.toBe(observerB.next.mock.calls[0][0]);
  });

  it('should handle late subscribers correctly', () => {
    vi.useFakeTimers();

    const source$ = new Observable(observer => {
      let i = 0;
      const intervalId = setInterval(() => {
        observer.next(i++);
      }, 1000);
      return () => clearInterval(intervalId);
    });

    const shared$ = source$.pipe(share());

    const observerA = { next: vi.fn() };
    const observerB = { next: vi.fn() };

    shared$.subscribe(observerA); // Subscribes at t=0

    vi.advanceTimersByTime(1500); // Emits 0, 1

    shared$.subscribe(observerB); // Subscribes at t=1.5s

    vi.advanceTimersByTime(1000); // Emits 2

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

    // ObserverB is a late subscriber and misses the first two emissions.
    expect(observerB.next).not.toHaveBeenCalledWith(0);
    expect(observerB.next).not.toHaveBeenCalledWith(1);
    expect(observerB.next).toHaveBeenCalledWith(2);

    vi.useRealTimers();
  });
});

Run your tests. This suite validates that your share operator correctly multicasts values, cleans up after itself, and re-subscribes when a new subscription cycle begins.

Conclusion

Excellent work! You have now implemented the share operator, a critical tool for optimizing reactive code. By combining a Subject with a reference counting mechanism, you've turned inefficient cold observables into efficient hot ones.

Key Takeaways:

• Cold vs. Hot: Cold observables create a new execution for each subscriber, while hot observables share a single execution.
• share Operator: Converts a cold, unicast observable into a hot, multicast one.
• Internal Mechanics: The share operator works by subscribing to the source once, multicasting values via an internal Subject, and using reference counting to automatically manage the lifecycle of the source subscription.

Preview of the Next Lesson:
With our core architecture (Observable, Subject, pipe) and a key multicasting operator (share) in place, we are now ready to build out our library's operator collection. In the next lesson, we will start with one of the most common and useful categories: transformation operators. We will implement a type-safe map operator to transform values as they flow through the stream.