Creating Observables with Static Factory Functions

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

In our last lesson, we equipped our Observable class with a powerful, type-safe pipe method. This enables us to compose operators, but we're missing two key things: the operators themselves, and easy ways to create observables to begin with.

Today, we'll address the second point. The goal for this lesson is to implement and test static factory functions (of, from, interval) for creating observables, ensuring correct type inference. These functions, often called "creation operators," are the entry point into the reactive world, allowing us to create observable streams from various sources like static values, iterables, and timers.

1. The Core Pattern: The Observable Producer

Before we build our factories, let's revisit the fundamental mechanism for creating any observable in our library: the Observable constructor. It takes a single argument, which we've called the producer function. This function encapsulates the entire logic of the observable: what values to emit, when to emit them, and how to handle errors or completion.

This producer function is lazy—it doesn't execute until subscribe is called. This "cold observable" behavior is key to making our factories work predictably.

To see this pattern in action, let's watch a video that demonstrates creating observables from scratch. It uses RxJS, but the core concept of a producer function passed to a constructor is identical to our implementation.

As you saw, the producer function is where all the magic happens. Our factory functions will be simple wrappers that return a new Observable with a pre-defined producer for a specific purpose.

Let's create a new file, src/factories.ts, where we'll place all our new factory functions. Make sure to import the Observable class.

// src/factories.ts
import { Observable } from './observable';

2. The of Factory: Emitting a Sequence of Values

The of factory is one of the simplest. It creates an observable that takes a list of arguments, emits each one in sequence, and then immediately completes.

This is useful for creating observables from a known, finite set of values, for example, for testing or for streams that start with some initial state.

Now, let's implement it.

Your Task: Implement and Test of

1. Add the of function to src/factories.ts:

   // src/factories.ts
   import { Observable } from './observable';
   
   /**
    * Creates an Observable that emits the arguments you provide and then completes.
    */
   export function of<T>(...args: T[]): Observable<T> {
     return new Observable<T>(observer => {
       for (const arg of args) {
         observer.next(arg);
       }
       observer.complete();
     });
   }
   

   • Type Inference: The use of a rest parameter ...args: T[] allows TypeScript to infer T as the most appropriate type. For of(1, 2, 3), T is number. For of(1, 'a', true), T becomes the union type string | number | boolean.

2. Create a test file src/factories.test.ts and add a test for of:

   // src/factories.test.ts
   import { describe, it, expect, vi } from 'vitest';
   import { of } from './factories';
   
   describe('of', () => {
     it('should emit a sequence of values and then complete', () => {
       const values: number[] = [];
       const nextFn = vi.fn(value => values.push(value));
       const completeFn = vi.fn();
   
       const source$ = of(1, 2, 3);
   
       source$.subscribe({
         next: nextFn,
         complete: completeFn,
       });
   
       // Assertions
       expect(nextFn).toHaveBeenCalledTimes(3);
       expect(completeFn).toHaveBeenCalledTimes(1);
       expect(values).toEqual([1, 2, 3]);
     });
   });
   

3. The from Factory: Converting Iterables

The from factory is more versatile. It converts an array, an iterable (like a string, Map, or Set), or a Promise into an observable. We'll focus on the iterable case.

Your Task: Implement and Test from

1. Add the from function to src/factories.ts:

   // src/factories.ts
   
   // ... existing 'of' function
   
   /**
    * Creates an Observable from an Array, a Promise, or an iterable.
    */
   export function from<T>(input: Iterable<T>): Observable<T> {
     return new Observable<T>(observer => {
       for (const value of input) {
         observer.next(value);
       }
       observer.complete();
     });
   }
   

   • Type Inference: By typing the input as Iterable<T>, TypeScript can infer T from the type of the elements within the iterable. For example, if you pass a string[], T becomes string.

2. Add tests for from in src/factories.test.ts:

   // src/factories.test.ts
   import { describe, it, expect, vi } from 'vitest';
   import { of, from } from './factories'; // update import
   
   // ... 'of' describe block
   
   describe('from', () => {
     it('should create an observable from an array', () => {
       const values: number[] = [];
       const source$ = from([10, 20, 30]);
   
       source$.subscribe(v => values.push(v));
   
       expect(values).toEqual([10, 20, 30]);
     });
   
     it('should create an observable from a string', () => {
       const values: string[] = [];
       const source$ = from('abc');
   
       source$.subscribe(v => values.push(v));
   
       expect(values).toEqual(['a', 'b', 'c']);
     });
   });
   

4. The interval Factory: Emitting Over Time

Now for our first truly asynchronous factory. interval creates an observable that emits an incrementing number at a specified time interval. Given your background in radiophysics, you can think of this as a digital clock signal generator, producing a new value at a regular period.

This implementation introduces a critical concept: teardown logic. Since setInterval will run forever, we must clean it up when a consumer unsubscribes to prevent memory leaks. Our Producer function can return a function, and our Subscription logic (from a previous lesson) will execute this returned function upon unsubscription.

Your Task: Implement and Test interval

1. Add the interval function to src/factories.ts:

   // src/factories.ts
   
   // ... existing functions
   
   /**
    * Creates an Observable that emits sequential numbers every specified
    * interval of time.
    */
   export function interval(period: number): Observable<number> {
     return new Observable<number>(observer => {
       let count = 0;
       const intervalId = setInterval(() => {
         observer.next(count);
         count += 1;
       }, period);
   
       // Return the teardown logic
       return () => {
         clearInterval(intervalId);
       };
     });
   }
   

2. Test interval using Vitest's fake timers. This allows us to test time-based logic deterministically without waiting. Add these tests to src/factories.test.ts.

   // src/factories.test.ts
   import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; // update import
   import { of, from, interval } from './factories'; // update import
   
   // ... existing describe blocks
   
   describe('interval', () => {
     beforeEach(() => {
       vi.useFakeTimers();
     });
   
     afterEach(() => {
       vi.useRealTimers();
     });
   
     it('should emit values at a specified period', () => {
       const nextFn = vi.fn();
       const source$ = interval(1000); // emit every 1s
   
       source$.subscribe({ next: nextFn });
   
       // At time 0, nothing has been emitted yet
       expect(nextFn).not.toHaveBeenCalled();
   
       // Advance time by 1s
       vi.advanceTimeBy(1000);
       expect(nextFn).toHaveBeenCalledTimes(1);
       expect(nextFn).toHaveBeenLastCalledWith(0);
   
       // Advance time by another 2s
       vi.advanceTimeBy(2000);
       expect(nextFn).toHaveBeenCalledTimes(3); // 1 (at 1s) + 2 (at 2s, 3s)
       expect(nextFn).toHaveBeenLastCalledWith(2);
     });
   
     it('should stop emitting values after unsubscription', () => {
       const nextFn = vi.fn();
       const source$ = interval(500);
   
       const subscription = source$.subscribe({ next: nextFn });
   
       // Advance time by 1s (emits 0, 1)
       vi.advanceTimeBy(1000);
       expect(nextFn).toHaveBeenCalledTimes(2);
   
       // Unsubscribe and advance time again
       subscription.unsubscribe();
       vi.advanceTimeBy(1000);
   
       // No new emissions should have occurred
       expect(nextFn).toHaveBeenCalledTimes(2);
     });
   });
   

Conclusion

Excellent work! You've just implemented three of the most fundamental observable creation functions. These factories are the primary way users will start interacting with your library.

Key Takeaways:

• Factory Functions are standalone functions that simplify the creation of observables by wrapping the new Observable(producer) pattern.
• of creates an observable from a sequence of arguments, emitting them synchronously and then completing.
• from creates an observable from an iterable (like an array or string), also emitting synchronously.
• interval creates an asynchronous observable that emits numbers over time. This highlighted the importance of teardown logic (returning a cleanup function from the producer) to prevent resource leaks.
• Fake timers in Vitest are an essential tool for testing asynchronous, time-based code reliably and quickly.

Preview of the Next Lesson:
The observables we've created so far are "cold" and "unicast"—each subscription gets its own independent execution of the producer. Next, we will implement a Subject. A Subject is a special hybrid that acts as both an Observer and an Observable, allowing it to multicast a single stream of values to multiple subscribers. This will open the door to creating "hot" observables and sharing data streams between different parts of an application.