useExtractedObservable

Category

@RxJS

Export Size

323 B

Package

@reause/rxjs

Last Changed

5 days ago

Use an RxJS Observable as extracted from one or more hooks, return the latest emitted value, and automatically unsubscribe from it when the component is unmounted.

Automatically unsubscribe on observable change, and automatically unsubscribe from it when the component is unmounted.

The source is a plain value, so the extractor re-runs when its identity changes; deps covers a source that is mutated in place.

Demo

Install

npm i rxjs

Usage

import { useExtractedObservable } from '@reause/rxjs'
import { useState } from 'react'
import { interval } from 'rxjs'
import { mapTo, scan, startWith } from 'rxjs/operators'

export function Counter() {
  const [start, setStart] = useState(0)

  const count = useExtractedObservable(start, start => interval(1000).pipe(
    mapTo(1),
    startWith(start),
    scan((total, next) => next + total),
  ))

  return (
    <div>
      <p>
        Counter:
        {count}
      </p>
      <button onClick={() => setStart(0)}>Restart from 0</button>
    </div>
  )
}

The subscription is created in an effect: it is unsubscribed whenever the source value changes, and on unmount. Upstream's watch options have no React equivalent — the extractor always runs on mount (upstream's immediate: true default), and a source object mutated in place is re-extracted by listing the mutation inputs in deps:

import { useExtractedObservable } from '@reause/rxjs'
import { useState } from 'react'
import { of } from 'rxjs'

const [filters, setFilters] = useState({ status: 'open', limit: 10 })

const label = useExtractedObservable(
  filters,
  filters => of(`${filters.status}: ${filters.limit}`),
  { deps: [filters.status, filters.limit] },
)

If you want to add custom error handling to an Observable that might error, you can supply an optional onError configuration. Without this, RxJS will treat any error in the supplied Observable as an "unhandled error" and it will be thrown in a new call stack and reported to window.onerror (or process.on('error') if you happen to be in Node).

import { useExtractedObservable } from '@reause/rxjs'
import { useState } from 'react'
import { interval } from 'rxjs'
import { mapTo, scan, startWith, tap } from 'rxjs/operators'

const [start, setStart] = useState(0)

const count = useExtractedObservable(
  start,
  (start) => {
    return interval(1000).pipe(
      mapTo(1),
      startWith(start),
      scan((total, next) => next + total),
      tap((n) => {
        if (n === 10)
          throw new Error('oops')
      }),
    )
  },
  {
    onError: (err: unknown) => {
      console.log(err) // Error: oops
    },
  },
)

You can also supply an optional onComplete configuration if you need to attach special behavior when the watched observable completes.

import { useExtractedObservable } from '@reause/rxjs'
import { useState } from 'react'
import { interval } from 'rxjs'
import { mapTo, scan, startWith, takeWhile } from 'rxjs/operators'

const [start, setStart] = useState(0)

const count = useExtractedObservable(
  start,
  (start) => {
    return interval(1000).pipe(
      mapTo(1),
      startWith(start),
      scan((total, next) => next + total),
      takeWhile(num => num < 10),
    )
  },
  {
    initialValue: 0,
    onComplete: () => {
      console.log('Done!')
    },
  },
)

Options

Option	Type	Description
initialValue	T	Value to use before the Observable emits
onError	(err: unknown) => void	Error handler for Observable errors
onComplete	() => void	Called when the Observable completes
deps	unknown[]	Extra dependencies that re-extract

Return Value

Returns the latest value emitted by the extracted Observable — a plain value instead of upstream's readonly ShallowRef:

const count = useExtractedObservable(start, start => interval(1000).pipe(
  startWith(start),
  scan((total, next) => next + total),
))

// `undefined` until the first emission, unless `initialValue` was provided
console.log(count)

Type Declarations

export interface UseExtractedObservableOptions<E> extends UseObservableOptions<E> {
    onComplete?: () => void;
    deps?: unknown[];
}

export type ExtractedObservableExtractor<Value, E> = (value: NonNullable<Value>, onCleanup: OnCleanup) => Observable<E>

export interface UseObservableOptions<I> {
    onError?: (err: unknown) => void;
    initialValue?: I | undefined;
}

export type OnCleanup = (cleanupFn: () => void) => void

export type WatchExtractedObservableExtractor<Value, E> = (value: NonNullable<Value>, onCleanup: OnCleanup) => Observable<E>

export function useExtractedObservable<Value, E, I = undefined>(value: Value | null | undefined, extractor: ExtractedObservableExtractor<Value, E>, options?: UseExtractedObservableOptions<E | I>): E | I

export function useObservable<H, I = undefined>(observable: Observable<H>, options?: UseObservableOptions<I | undefined>): UseObservableReturn<H, I>

Source

Source · Demo

Contributors

Changelog

v0.1.0 on

19b24 - feat(rxjs): implement useExtractedObservable (#125)