RxJS Observables Explained: Create from Arrays, Strings, Events with of, from and fromEvent

The earlier RxJS lesson explained what an Observable is. Here you learn the creation functions: the many ways to start a stream from strings, arrays, objects, collections, promises, timers and DOM events.

What you will learn

• The Observer: next, error, complete
• Creation functions: of, from, range, interval, timer, defer, new Observable
• fromEvent for browser events
• pipe() with map and filter

The Observer

When you subscribe, you can pass an observer with three handlers:

import { of } from 'rxjs';

of(1, 2, 3).subscribe({
  next: (value) => console.log('value', value), // each item
  error: (err) => console.error('error', err), // a failure (ends the stream)
  complete: () => console.log('done'), // finished (ends the stream)
});
// value 1, value 2, value 3, done

A stream sends zero or more next values, then ends with either error or complete, never both. After that, nothing more arrives.

Creating Observables

// src/app/rxjs2/creation.ts
import { defer, from, fromEvent, interval, Observable, of, range, timer } from 'rxjs';

// of: emits each argument. Works with strings, numbers, objects, anything.
export const letters$ = of('a', 'b', 'c');
export const objects$ = of({ id: 1 }, { id: 2 });
export const arrayAsOne$ = of([1, 2, 3]); // emits ONE value: the array

// from: converts something iterable into a stream of its items
export const fromArray$ = from([10, 20, 30]); // 10, 20, 30
export const fromString$ = from('abc'); // 'a', 'b', 'c'
export const fromSet$ = from(new Set([1, 2, 2, 3])); // 1, 2, 3
export const fromMap$ = from(new Map([['x', 1], ['y', 2]])); // ['x', 1], ['y', 2]
export const fromObjectEntries$ = from(Object.entries({ a: 1, b: 2 })); // ['a', 1], ['b', 2]
export const fromPromise$ = from(Promise.resolve('from a promise')); // one value

// range: a sequence of numbers
export const range$ = range(5, 3); // 5, 6, 7

// timers
export const everySecond$ = interval(1000); // 0, 1, 2, ... forever
export const once$ = timer(500); // emits 0 after 500 ms, then completes
export const delayedThenEvery$ = timer(500, 1000); // after 500 ms, then every second

// defer: create the source when someone subscribes (fresh every time)
export const now$ = defer(() => of(new Date().toISOString()));

// new Observable: full control. The function runs on every subscribe.
export const custom$ = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  const id = setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
  }, 100);
  // The returned function is the teardown, called on unsubscribe or completion
  return () => clearTimeout(id);
});

// fromEvent: browser events as a stream
export function clicks(el: HTMLElement): Observable<Event> {
  return fromEvent(el, 'click');
}

Key differences:

• of([1, 2, 3]) emits one value (the array). from([1, 2, 3]) emits three values.
• from(promise) turns a promise into an Observable, and firstValueFrom(observable) or lastValueFrom(observable) turn an Observable back into a promise.
• interval never completes. You must stop it (take, takeUntil, or unsubscribe).
• Observables are lazy and cold: nothing runs until subscribe, and every subscriber gets its own run. defer is handy when the source depends on the time of subscription.

fromEvent: events as streams

// src/app/rxjs2/click-counter.ts
import { Component, ElementRef, afterNextRender, inject, signal } from '@angular/core';
import { DestroyRef } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { filter, fromEvent, map, scan } from 'rxjs';

@Component({
  selector: 'app-click-counter',
  template: `
    <button #btn type="button">Click me (only Shift+Click counts)</button>
    <p>Counted: {{ count() }}</p>
    <p>Last key: {{ lastKey() }}</p>
  `,
})
export class ClickCounter {
  private readonly host = inject<ElementRef<HTMLElement>>(ElementRef);
  private readonly destroyRef = inject(DestroyRef);
  protected readonly count = signal(0);
  protected readonly lastKey = signal('none');

  constructor() {
    afterNextRender(() => {
      const button = this.host.nativeElement.querySelector('button')!;

      fromEvent<MouseEvent>(button, 'click')
        .pipe(
          filter((e) => e.shiftKey), // keep only shift-clicks
          scan((n) => n + 1, 0), // running total
          takeUntilDestroyed(this.destroyRef),
        )
        .subscribe((n) => this.count.set(n));

      fromEvent<KeyboardEvent>(window, 'keydown')
        .pipe(map((e) => e.key), takeUntilDestroyed(this.destroyRef))
        .subscribe((key) => this.lastKey.set(key));
    });
  }
}

fromEvent adds the DOM listener when you subscribe and removes it when you unsubscribe. The afterNextRender wrapper keeps the code browser-only and waits until the button exists. Often the template (click) is simpler, but fromEvent shines when you need operators on the events: debouncing, filtering, combining.

pipe, map and filter

pipe() chains operators. Each operator takes an Observable and returns a new one, and the original is never changed:

import { from, map, filter } from 'rxjs';

from([1, 2, 3, 4, 5, 6])
  .pipe(
    filter((n) => n % 2 === 0), // 2, 4, 6
    map((n) => n * n), // 4, 16, 36
  )
  .subscribe(console.log);

• map transforms each value.
• filter keeps values that pass a test.

Operators are described by what happens to each value and, for time-based ones, when. Nothing runs until you subscribe.

Test the behaviour

// src/app/rxjs2/creation.spec.ts
import { fromEvent, from, of, take, map, filter, toArray, firstValueFrom, lastValueFrom, timer, interval } from 'rxjs';
import { arrayAsOne$, fromArray$, fromMap$, fromObjectEntries$, fromSet$, fromString$, letters$, range$ } from './creation';

const collect = <T>(source: import('rxjs').Observable<T>) => lastValueFrom(source.pipe(toArray()));

describe('creating observables', () => {
  it('of emits each argument, from emits each item', async () => {
    expect(await collect(letters$)).toEqual(['a', 'b', 'c']);
    expect(await collect(arrayAsOne$)).toEqual([[1, 2, 3]]);
    expect(await collect(fromArray$)).toEqual([10, 20, 30]);
    expect(await collect(fromString$)).toEqual(['a', 'b', 'c']);
    expect(await collect(fromSet$)).toEqual([1, 2, 3]);
    expect(await collect(fromMap$)).toEqual([['x', 1], ['y', 2]]);
    expect(await collect(fromObjectEntries$)).toEqual([['a', 1], ['b', 2]]);
    expect(await collect(range$)).toEqual([5, 6, 7]);
  });

  it('pipe chains map and filter', async () => {
    const result = await collect(from([1, 2, 3, 4, 5, 6]).pipe(filter((n) => n % 2 === 0), map((n) => n * n)));
    expect(result).toEqual([4, 16, 36]);
  });

  it('subscribe calls next, then complete', () => {
    const events: string[] = [];
    of(1, 2).subscribe({ next: (v) => events.push('next' + v), complete: () => events.push('complete') });
    expect(events).toEqual(['next1', 'next2', 'complete']);
  });

  it('fromEvent listens to DOM events and removes the listener on unsubscribe', () => {
    const button = document.createElement('button');
    const seen: string[] = [];
    const sub = fromEvent(button, 'click').subscribe(() => seen.push('click'));
    button.click();
    sub.unsubscribe();
    button.click();
    expect(seen).toEqual(['click']);
  });

  it('firstValueFrom turns an observable into a promise', async () => {
    expect(await firstValueFrom(of('first', 'second'))).toBe('first');
  });

  it('interval and timer work with fake timers', () => {
    vi.useFakeTimers();
    const ticks: number[] = [];
    interval(1000).pipe(take(3)).subscribe((n) => ticks.push(n));
    vi.advanceTimersByTime(3500);
    expect(ticks).toEqual([0, 1, 2]);

    let fired = false;
    timer(500).subscribe(() => (fired = true));
    vi.advanceTimersByTime(499);
    expect(fired).toBe(false);
    vi.advanceTimersByTime(2);
    expect(fired).toBe(true);
    vi.useRealTimers();
  });
});

Put the click counter on the page:

// src/app/app.ts
import { Component } from '@angular/core';
import { ClickCounter } from './rxjs2/click-counter';

@Component({
  imports: [ClickCounter],
  selector: 'app-root',
  templateUrl: './app.html',
  styleUrl: './app.css',
})
export class App {}

<!-- src/app/app.html -->
<main>
  <h1>fromEvent</h1>
  <app-click-counter />
</main>

Try it yourself

1. Create an Observable from an object's values with from(Object.values(obj)).
2. Use fromEvent on window resize and show the width with map.
3. Wrap setInterval yourself in new Observable, including the teardown, and test it.

Quick check

1. What is the difference between of([1, 2]) and from([1, 2])?
2. When does an Observable start running?
3. What does the function returned from new Observable(...) do?

Next, control how many values you take or skip in filtering and utility operators. See the RxJS operators list.