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

refactor: remove nested subscription (#36)

Browse Source
This commit is contained in:
Alex Inkin
2022-06-23 00:23:35 +03:00
committed by GitHub
parent e138b88e75
commit 9a28793f27
1 changed files with 17 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,4 +1,5 @@
import { merge, Observable, ReplaySubject, Subject, Subscription } from 'rxjs' import { merge, Observable, ReplaySubject, Subject } from 'rxjs'
import { 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'
@@ -10,15 +11,14 @@ export class PatchDB<T> {
public rpcError$ = new Subject<RPCError>() public rpcError$ = new Subject<RPCError>()
public cache$ = new ReplaySubject<DBCache<T>>(1) public cache$ = new ReplaySubject<DBCache<T>>(1)
private updatesSub?: Subscription private sub = this.sources$.pipe(
private sourcesSub = this.sources$.subscribe(sources => { switchMap(sources => merge(...sources.map(s => s.watch$(this.store)))),
this.updatesSub = merge(...sources.map(s => s.watch$(this.store))).subscribe({ ).subscribe({
next: (res) => { next: (res) => {
if ('result' in res) { if ('result' in res) {
this.store.update(res.result) this.store.update(res.result)
this.cache$.next(this.store.cache) this.cache$.next(this.store.cache)
} } else {
else {
this.rpcError$.next(res) this.rpcError$.next(res)
} }
}, },
@@ -26,7 +26,6 @@ export class PatchDB<T> {
this.connectionError$.next(e) this.connectionError$.next(e)
}, },
}) })
})
constructor ( constructor (
private readonly sources$: Observable<Source<T>[]>, private readonly sources$: Observable<Source<T>[]>,
@@ -35,7 +34,6 @@ export class PatchDB<T> {
) { } ) { }
clean () { clean () {
this.sourcesSub.unsubscribe() this.sub.unsubscribe()
this.updatesSub?.unsubscribe()
} }
} }