Drop-in replacements for RxJS Observable methods and operators that work with AbortSignal
npm install abortable-rx
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 create
Easiest way to wrap an abortable async function into a Promise. The factory is called every time the Observable is subscribed to, and the AbortSignal is aborted on unsubscription.
-
Creates an Observable just like RxJS , but exposes an AbortSignal in addition to the subscriber
- toPromise
Returns a Promise that resolves with the last emission of the given Observable, rejects if the Observable errors or rejects with an when the AbortSignal is aborted.forEach
-
Calls 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
Like RxJS , but passes an AbortSignal that is aborted when the source emits another element.concatMap
-
Like RxJS , but passes an AbortSignal that is aborted when the returned Observable is unsubscribed from.mergeMap
-
Like RxJS , 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.finally
This is so calling code does not continue execution and gets a chance to cleanup with .error.name === 'AbortError'
You can handle this error (usually at the top level) by checking if 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.
`ts
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)
`ts
import { toPromise } from 'abortable-rx'
class ClientConnection {
private events: Observable
async sync(signal?: AbortSignal): Promise
await this.scheduleSync('immediatly', signal)
const stream = this.events.pipe(
filter(event => event.type === 'SYNC_COMPLETED'),
take(1)
)
await toPromise(stream, signal)
}
}
`
`ts
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.