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

re-introduce watch

Browse Source
This commit is contained in:
Matt Hill
2021-07-07 15:28:52 -06:00
committed by Aiden McClelland
parent 434c6f7037
commit 0cbdc781b0
1 changed files with 32 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
client/lib/store.ts
View File

@@ -1,11 +1,12 @@
import { DBCache, Dump, Revision, Update } from './types' import { DBCache, Dump, Revision, Update } from './types'
import { applyPatch } from 'fast-json-patch' import { applyPatch, getValueByPointer } from 'fast-json-patch'
import { BehaviorSubject, from, Observable } from 'rxjs' import { BehaviorSubject, Observable } from 'rxjs'
import { toStream } from 'mobx-utils' import { finalize } from 'rxjs/operators'
export class Store<T> { export class Store<T> {
cache: DBCache<T> cache: DBCache<T>
sequence$: BehaviorSubject<number> sequence$: BehaviorSubject<number>
private nodes: { [path: string]: BehaviorSubject<any> } = { }
constructor ( constructor (
readonly initialCache: DBCache<T>, readonly initialCache: DBCache<T>,
@@ -22,30 +23,47 @@ export class Store<T> {
watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5): Observable<T[P1][P2][P3][P4][P5]> watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5): Observable<T[P1][P2][P3][P4][P5]>
watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4], P6 extends keyof T[P1][P2][P3][P4][P5]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5, p6: P6): Observable<T[P1][P2][P3][P4][P5][P6]> watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4], P6 extends keyof T[P1][P2][P3][P4][P5]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5, p6: P6): Observable<T[P1][P2][P3][P4][P5][P6]>
watch$ (...args: (string | number)[]): Observable<any> { watch$ (...args: (string | number)[]): Observable<any> {
return from(toStream(() => this.peekNode(...args), true)) const path = `/${args.join('/')}`
if (!this.nodes[path]) {
this.nodes[path] = new BehaviorSubject(getValueByPointer(this.cache.data, path))
this.nodes[path].pipe(
finalize(() => delete this.nodes[path]),
)
}
return this.nodes[path].asObservable()
} }
update (update: Update<T>): void { update (update: Update<T>): void {
if ((update as Revision).patch) { if ((update as Revision).patch) {
if (this.cache.sequence + 1 !== update.id) throw new Error(`Outdated sequence: current: ${this.cache.sequence}, new: ${update.id}`) if (this.cache.sequence + 1 !== update.id) throw new Error(`Outdated sequence: current: ${this.cache.sequence}, new: ${update.id}`)
applyPatch(this.cache.data, (update as Revision).patch, true, true) applyPatch(this.cache.data, (update as Revision).patch, true, true);
(update as Revision).patch.forEach(op => {
this.updateNodesByPath(op.path)
})
} else { } else {
this.cache.data = (update as Dump<T>).value this.cache.data = (update as Dump<T>).value
this.updateNodesByPath('')
} }
this.cache.sequence = update.id this.cache.sequence = update.id
this.sequence$.next(this.cache.sequence) this.sequence$.next(this.cache.sequence)
} }
reset (): void { updateNodesByPath (revisionPath: string) {
this.cache.sequence = 0 Object.keys(this.nodes).forEach(nodePath => {
this.cache.data = { } as T if (!this.nodes[nodePath]) return
if (nodePath.includes(revisionPath) || revisionPath.includes(nodePath)) {
try {
this.nodes[nodePath].next(getValueByPointer(this.cache.data, nodePath))
} catch (e) {
this.nodes[nodePath].complete()
delete this.nodes[nodePath]
}
}
})
} }
private peekNode (...args: (string | number)[]): any { reset (): void {
try { Object.values(this.nodes).forEach(node => node.complete())
return args.reduce((acc, next) => (acc as any)[`${next}`], this.cache.data)
} catch (e) {
return undefined
}
} }
} }