RxJS Interoperability with Adapters

Hello! Welcome back.

In our last lesson, you built a powerful expectObservable assertion helper, completing our reactive testing suite. This allows us to write clean, declarative, and deterministic tests for our asynchronous operators, which is a massive step forward for the quality and maintainability of our library.

Today, we'll shift our focus from internal tooling to external integration. For any new library to succeed, it must be able to coexist with the established ecosystem. In the world of reactive programming in JavaScript, that ecosystem is dominated by RxJS. This lesson directly addresses the learning outcome: Implement adapters to convert between your library's observables and RxJS observables.

We will build two small but critical utility functions:

1. fromRxJs: To convert an RxJS observable into one of our own.
2. toRxJs: To convert one of our observables into an RxJS observable.

These adapters will make our library dramatically more useful in real-world projects, allowing for gradual adoption and seamless interoperability.

1. The Principle of Interoperability

Why is this so important? Imagine a team wants to use a cool new operator from our library in a large, existing Angular project that is already heavily invested in RxJS. They can't just replace everything. They need a bridge. Our adapters will provide that bridge.

The good news is that this conversion is straightforward. Since our library was designed with RxJS principles in mind, the core concepts are identical. Both our Observable and the RxJS Observable rely on the same contract:

• A subscribe method that accepts an observer object.
• An observer object with next, error, and complete methods.
• A subscription object returned by subscribe that has an unsubscribe method for teardown.

This shared structure is the key. The conversion process is essentially an application of the Adapter design pattern: we wrap one object with an interface that another system expects.

Let's look at how this is done in a similar context.

The key insight is that you can create an observable of one type by subscribing to an observable of another type within the new observable's constructor. Now, let's implement this pattern for our library.

2. Implementing the toRxJs Adapter

Let's start by converting one of our observables into an RxJS observable. For this, you'll first need to add RxJS to your project as a development dependency.

npm install rxjs --save-dev

Now, we can create our toRxJs adapter. The logic is simple: we create a new Rx.Observable and, within its producer function, we subscribe to our source observable, proxying all notifications.

A critical piece is handling the teardown logic. When a consumer unsubscribes from the new RxJS observable, we must ensure the subscription to our original observable is also terminated to prevent memory leaks.

// src/interop/toRxJs.ts

import { Observable as RxObservable } from 'rxjs';
import { Observable } from '../core/Observable';

/**
 * Converts one of our Observables to an RxJS Observable.
 * @param source Our custom observable instance.
 * @returns An RxJS Observable that mirrors the source.
 */
export function toRxJs<T>(source: Observable<T>): RxObservable<T> {
  // 1. Create a new RxJS Observable. Its constructor receives a subscriber.
  return new RxObservable<T>(subscriber => {
    // 2. Subscribe to our source observable.
    const subscription = source.subscribe({
      // 3. Proxy notifications from our source to the RxJS subscriber.
      next: value => subscriber.next(value),
      error: err => subscriber.error(err),
      complete: () => subscriber.complete(),
    });

    // 4. Return the teardown logic. When the RxJS observable is unsubscribed,
    //    this will unsubscribe from our source observable.
    return () => {
      subscription.unsubscribe();
    };
  });
}

This is the exact pattern highlighted in the Stack Overflow resource and is fundamental to creating robust adapters. The resource below explains this teardown mechanism in great detail.

3. Implementing the fromRxJs Adapter

Now for the other direction: converting an RxJS observable into one of our own. The logic is a mirror image of what we just did.

// src/interop/fromRxJs.ts

import { Observable as RxObservable } from 'rxjs';
import { Observable } from '../core/Observable';

/**
 * Converts an RxJS Observable to one of our Observables.
 * @param source An RxJS Observable instance.
 * @returns Our custom observable that mirrors the source.
 */
export function fromRxJs<T>(source: RxObservable<T>): Observable<T> {
  // 1. Create a new instance of our Observable. Its constructor receives an observer.
  return new Observable<T>(observer => {
    // 2. Subscribe to the RxJS source observable.
    const subscription = source.subscribe({
      // 3. Proxy notifications from the RxJS source to our observer.
      next: value => observer.next(value),
      error: err => observer.error(err),
      complete: () => observer.complete(),
    });

    // 4. Return our teardown logic. The object returned by our producer
    //    must have an `unsubscribe` method.
    return {
      unsubscribe: () => {
        subscription.unsubscribe();
      },
    };
  });
}

Notice the symmetry between the two functions. They both act as a thin wrapper, translating the subscription and notification flow from one implementation to the other.

4. Testing the Adapters

Now that we have our adapters, let's prove they work. We can use the testing infrastructure we've built.

First, let's test fromRxJs. We can create a source observable using RxJS's of and delay operators, convert it using our adapter, and then assert its behavior using our own TestScheduler.

// src/interop/fromRxJs.test.ts

import { describe, it, expect } from 'vitest';
import { TestScheduler } from '../testing/TestScheduler';
import { of as rxOf, delay as rxDelay } from 'rxjs';
import { fromRxJs } from './fromRxJs';

describe('fromRxJs', () => {
  it('should convert an RxJS observable to our observable', () => {
    const scheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });

    scheduler.run(({ expectObservable }) => {
      // Create an RxJS observable that emits 'a' after 20ms
      const rxSource = rxOf('a').pipe(rxDelay(20, scheduler as any));
      
      // Convert it to our observable
      const ourStream = fromRxJs(rxSource);

      // Assert its behavior using our tools
      const expectedMarbles = '--(a|)';
      expectObservable(ourStream).toBe(expectedMarbles, { a: 'a' });
    });
  });
});

Note: We cast scheduler as any because the RxJS delay operator expects an Rx.SchedulerLike, and while our scheduler is compatible in this testing context, TypeScript can't prove it. A more advanced integration might involve creating a dedicated scheduler adapter.

Testing toRxJs requires us to use RxJS's testing tools to verify the output. This is a great exercise and a direct preview of our next lesson.

// src/interop/toRxJs.test.ts

import { describe, it } from 'vitest';
import { TestScheduler as RxTestScheduler } from 'rxjs/testing';
import { of } from '../creators/of';
import { delay } from '../operators/delay';
import { toRxJs } from './toRxJs';

describe('toRxJs', () => {
  it('should convert our observable to an RxJS observable', () => {
    // 1. Create an RxJS TestScheduler
    const rxScheduler = new RxTestScheduler((actual, expected) => {
      // Use vitest's expect for the assertion
      expect(actual).toEqual(expected);
    });

    // 2. Run the test within the RxJS scheduler's context
    rxScheduler.run(({ cold, expectObservable }) => {
      // Create one of our observables using its `cold` helper
      const ourSource = of('a');
      const ourDelayedStream = ourSource.pipe(delay(20, rxScheduler as any));
      
      // Convert it to an RxJS observable
      const rxStream = toRxJs(ourDelayedStream);

      // Assert its behavior using RxJS's expectObservable
      const expectedMarbles = '--(a|)';
      expectObservable(rxStream).toBe(expectedMarbles, { a: 'a' });
    });
  });
});

This test confirms that our toRxJs adapter correctly translates our observable's behavior into the RxJS world.

Conclusion

Fantastic work. You've just built a vital bridge between your library and the broader reactive ecosystem. By creating toRxJs and fromRxJs adapters, you've made your library exponentially more practical for real-world use cases.

Key Takeaways:

• Interoperability is Key: For a library to be adopted, it must play well with others. Adapters are the mechanism to achieve this.
• Shared Contract: The conversion is made possible by the fact that our observable and RxJS observables share the same fundamental subscription contract (next, error, complete).
• Adapter Pattern: We implemented the adapter pattern by creating a new observable of the target type and proxying notifications from the source observable within its constructor.
• Teardown is Crucial: Correctly managing the teardown logic by unsubscribing from the source when the wrapper is unsubscribed is essential to prevent memory leaks.

In our next lesson, we will leverage these adapters to write integration tests to verify compatibility with core RxJS operators. Having these adapters and understanding how to test them is the perfect foundation for that task.