Drop-in replacements for RxJS Observable methods and operators that work with AbortSignal
MIT License
Drop-in replacements for RxJS Observable methods and operators that work with AbortSignal
.
Enables easy interop between Observable code and Promise-returning functions, without losing the cancellation capabilities of RxJS.
Some operations are imperative by nature and easier to express in imperative code with async/await.
Expressing these operations that need control flow with functional RxJS operators or Subjects results in unreadable and unmaintainable code.
In addition, it is confusing to have async functions that only have one or no result, but return an Observable, as it is unclear how many times it will emit.
RxJS has great interop with Promises, however, it doesn't provide an easy mechanism to propagate cancellation to promise-returning functions like the native fetch
API.
This micro library provides that mechanism.
npm install abortable-rx
defer<T>(factory: (signal: AbortSignal) => ObservableInput<T>): Observable<T>
create<T>(subscribe?: (subscriber: Subscriber<T>, signal: AbortSignal) => TeardownLogic): Observable<T>
create
, but exposes an AbortSignal in addition to the subscribertoPromise<T>(observable: Observable<T>, signal?: AbortSignal): Promise<T>
AbortError
when the AbortSignal is aborted.forEach<T>(source: Observable<T>, next: (value: T) => void, signal?: AbortSignal): Promise<void>
next
for every emission and returns a Promise that resolves when the Observable completed, rejects if the Observable errors or rejects with an AbortError
when the AbortSignal is aborted.switchMap<T, R>(project: (value: T, index: number, abortSignal: AbortSignal) => ObservableInput<R>): OperatorFunction<T, R>
switchMap
, but passes an AbortSignal that is aborted when the source emits another element.concatMap<T, R>(project: (value: T, index: number, abortSignal: AbortSignal) => ObservableInput<R>): OperatorFunction<T, R>
concatMap
, but passes an AbortSignal that is aborted when the returned Observable is unsubscribed from.mergeMap<T, R>(project: (value: T, index: number, abortSignal: AbortSignal) => ObservableInput<R>): OperatorFunction<T, R>
mergeMap
, but passes an AbortSignal that is aborted when the returned Observable is unsubscribed from.forEach
and toPromise
will reject the Promise with an Error
if the signal is aborted.
This is so calling code does not continue execution and gets a chance to cleanup with finally
.
You can handle this error (usually at the top level) by checking if error.name === 'AbortError'
in a catch
block.
If the functions you pass to defer
, switchMap
, etc. throw AbortError
, you don't have to worry about catching it.
The Promises are always converted to Observables internally, and that Observable is always unsubscribed from first, then the AbortSignal is aborted.
After an Observable is unsubscribed from, all further emissions or errors are ignored, so you don't have to worry about the error terminating your Observable chain.
fetch
inside switchMap
import { fromEvent } from 'rxjs'
import { switchMap } from 'abortable-rx/operators'
fromEvent(input, 'value')
.pipe(switchMap(async (event, i, signal) => {
const resp = await fetch(`api/suggestions?value=${event.target.value}`, { signal })
if (!resp.ok) {
throw new Error(resp.statusText)
}
return await resp.json()
})
.subscribe(displaySuggestions)
toPromise
to wait for an event to happen onceimport { toPromise } from 'abortable-rx'
class ClientConnection {
private events: Observable<Event>
async sync(signal?: AbortSignal): Promise<void> {
await this.scheduleSync('immediatly', signal)
const stream = this.events.pipe(
filter(event => event.type === 'SYNC_COMPLETED'),
take(1)
)
await toPromise(stream, signal)
}
}
import { fromEvent } from 'rxjs'
import { switchMap } from 'rxjs/operators'
import { defer } from 'abortable-rx'
fromEvent(repoDropdown, 'change')
.pipe(switchMap(event =>
concat(
['Loading...'],
defer(async signal => {
while (true) {
const resp = await fetch(`api/repo/${event.target.value}`, { signal })
if (!resp.ok) {
throw new Error(resp.statusText)
}
const repo = await resp.json()
if (repo.cloneInProgress) {
await new Promise(resolve => setTimeout(resolve, 1000))
continue
}
return repo.filesCount
}
})
)
}))
.subscribe(content => {
fileCount.textContent = content
})
AbortSignal
is supported by all modern browsers, but there is a polyfill available if you need it.