toObservable

TL;DR. toObservable(source) — the bridge from Synapse to RxJS. It turns a storage (optionally a slice of it), a selector, or any core stream (selector.$, dispatcher.action$, a watcher, synapse.state$) into an RxJS Observable. It's a low-level utility for effects and non-React code. Imported from synapse-storage/reactive (requires rxjs). The examples use the end-to-end todoStorage (TodoState = { todos: Todo[]; filter: Filter }).

Why

A storage is not an RxJS source on its own: it has subscribe/getStateSync, not pipe. As soon as you need to run the state through operators (debounceTime, scan, bufferTime, …) or feed it into createEffectConfig as external state, you need an Observable. toObservable builds exactly that bridge: it emits the current state on subscribe, then on every change.

When to use / when you DON'T need it

Use it:

You DON'T need it:

In a React component, do not create toObservable(...) directly in render — a new Observable on every render triggers re-subscriptions. Memoize it or pass it as a factory to useObservable.

Signature

// whole state
toObservable<T>(storage: IStorageBase<T>): Observable<T>

// slice + optional comparator
toObservable<T, R>(
  storage: IStorageBase<T>,
  selector: (state: T) => R,
  equals?: (a: R, b: R) => boolean,
): Observable<R>

// selector (SelectorAPI) → its value stream
toObservable<T>(selector: SelectorAPI<T>): Observable<T>

// any core stream: selector.$, dispatcher.action$, d.someWatcher(), synapse.state$
toObservable<T>(stream: Subscribable<T>): Observable<T>

The core streams are lightweight interop observables without pipe (the core doesn't depend on RxJS); toObservable is equivalent to rxjs from(stream):

toObservable(selectors.searchQuery).pipe(debounceTime(300))
toObservable(dispatcher.action$).pipe(ofType(dispatcher.loadPosts))
toObservable(coreSynapse.state$) // e.g. to pass into an Effects constructor

What a subscriber gets right away depends on the source — toObservable adds nothing and drops nothing:

That's why withLatestFrom(toObservable(otherSynapse.state$)) in an effect is safe: an action dispatched right after ready() already sees the current state of the other store.

selector — a slice instead of the whole state

Without a selector the stream emits the whole state on every store change — even when a field you don't care about changed. With a selector the stream maps the state to a slice and runs it through distinctUntilChanged, so it emits only when the slice actually changed:

import { toObservable } from 'synapse-storage/reactive'

const state$ = toObservable(todoStorage)                        // Observable<TodoState>, on any change
const count$ = toObservable(todoStorage, (s) => s.todos.length) // Observable<number>, only when it changes

Here count$ won't fire if filter changed — the todos length is the same. That's the optimization: the subscriber (component/effect) doesn't wake up on unrelated changes.

equals — how slices are compared

The third parameter is the comparator for distinctUntilChanged. It decides whether a new slice value is "the same" (return true → the emission is skipped). By default the comparison uses Object.is.

Object.is is enough for primitives (number, string, boolean) and for slices with a stable reference (when immutable updates don't touch that object — the reference is preserved). A custom equals is needed in two cases:

1. The selector returns a new object/array every time. Then Object.is sees a new reference on every tick and distinctUntilChanged won't de-dupe — the stream emits on every store change. Pass a by-content comparison:

// factory selector: a new array each time → needs a by-value equals
const ids$ = toObservable(
  todoStorage,
  (s) => s.todos.map((t) => t.id),
  (a, b) => a.length === b.length && a.every((id, i) => id === b[i]),
)

2. You need a coarser equivalence than identity. For example, emit only when a property of the value changes, not the value itself:

// emits only when the parity of the count changes (1→3 stays quiet, 3→4 emits)
const parity$ = toObservable(todoStorage, (s) => s.todos.length, (a, b) => a % 2 === b % 2)

equals only makes sense together with selector — without a slice there's nothing to compare.

In effects

A typical case is wiring one storage's state as an external state into createEffectConfig:

const auth$ = toObservable(authStorage, (s) => s.user.id)

createEffectConfig: () => ({
  externalStates: { userId: auth$ },
})

All parameters (commented)

import { toObservable } from 'synapse-storage/reactive'

const slice$ = toObservable(
  // 1. storage — IStorageBase (Memory/Local/IndexedDB — a shared interface).
  //    The stream subscribes to storage.subscribeToAll and emits getStateSync().
  todoStorage,

  // 2. selector? — which slice to pull out. Without it the stream emits the WHOLE state on any
  //    change. With it — map + distinctUntilChanged, emitting only when the slice changes.
  (s) => s.todos.length,

  // 3. equals? — comparator for distinctUntilChanged (defaults to Object.is).
  //    Needed if the selector returns a new object/array every tick, or when you want a
  //    coarser equivalence. Only makes sense TOGETHER with selector.
  (a, b) => a === b,
)

Parameters

Notes

See also