Start9/patch-db
2
0
Fork 0
mirror of https://github.com/Start9Labs/patch-db.git synced 2026-03-26 10:21:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: properly handle source errors without stopping the stream

Browse Source
This commit is contained in:
waterplea
2022-06-29 11:51:28 +03:00
committed by Matt Hill
parent 9a28793f27
commit d24bd1c513
1 changed files with 16 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
client/lib/patch-db.ts
View File

@@ -1,5 +1,5 @@
import { merge, Observable, ReplaySubject, Subject } from 'rxjs' import { EMPTY, merge, Observable, ReplaySubject, Subject } from 'rxjs'
import { switchMap } from 'rxjs/operators' import { catchError, switchMap } from 'rxjs/operators'
import { Store } from './store' import { Store } from './store'
import { DBCache, Http } from './types' import { DBCache, Http } from './types'
import { RPCError } from './source/ws-source' import { RPCError } from './source/ws-source'
@@ -12,19 +12,20 @@ export class PatchDB<T> {
public cache$ = new ReplaySubject<DBCache<T>>(1) public cache$ = new ReplaySubject<DBCache<T>>(1)
private sub = this.sources$.pipe( private sub = this.sources$.pipe(
switchMap(sources => merge(...sources.map(s => s.watch$(this.store)))), switchMap(sources => merge(...sources.map(s => s.watch$(this.store))).pipe(
).subscribe({ catchError(e => {
next: (res) => { this.connectionError$.next(e)
if ('result' in res) {
this.store.update(res.result) return EMPTY
this.cache$.next(this.store.cache) }),
} else { )),
this.rpcError$.next(res) ).subscribe((res) => {
} if ('result' in res) {
}, this.store.update(res.result)
error: (e) => { this.cache$.next(this.store.cache)
this.connectionError$.next(e) } else {
}, this.rpcError$.next(res)
}
}) })
constructor ( constructor (