mirror of
https://github.com/Start9Labs/patch-db.git
synced 2026-03-26 02:11:54 +00:00
feat: make watch$ strictNullCheck compatible (#34)
* feat: make `watch$` strictNullCheck compatible * chore: send initial value to subscribers * let values emit when undefined * fix: properly check if data can be processed Co-authored-by: Lucy Cifferello <12953208+elvece@users.noreply.github.com>
This commit is contained in:
@@ -75,7 +75,7 @@ function recursiveApply<T extends Record<string, T>> (data: T, path: readonly st
|
|||||||
}
|
}
|
||||||
|
|
||||||
function isObject (val: any): val is Record<string, unknown> {
|
function isObject (val: any): val is Record<string, unknown> {
|
||||||
return typeof val === 'object' && !Array.isArray(val) && val !== null
|
return typeof val === 'object' && val !== null
|
||||||
}
|
}
|
||||||
|
|
||||||
function jsonPathToKeyArray (path: string): string[] {
|
function jsonPathToKeyArray (path: string): string[] {
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { DBCache, Dump, Http, Revision, Update } from './types'
|
import { DBCache, Dump, Http, Revision, Update } from './types'
|
||||||
import { BehaviorSubject, Observable } from 'rxjs'
|
import { BehaviorSubject, Observable, ReplaySubject } from 'rxjs'
|
||||||
import { applyOperation, getValueByPointer, Operation } from './json-patch-lib'
|
import { applyOperation, getValueByPointer, Operation } from './json-patch-lib'
|
||||||
import BTree from 'sorted-btree'
|
import BTree from 'sorted-btree'
|
||||||
|
|
||||||
@@ -11,10 +11,10 @@ export interface StashEntry {
|
|||||||
export class Store<T extends { [key: string]: any }> {
|
export class Store<T extends { [key: string]: any }> {
|
||||||
cache: DBCache<T>
|
cache: DBCache<T>
|
||||||
sequence$: BehaviorSubject<number>
|
sequence$: BehaviorSubject<number>
|
||||||
private watchedNodes: { [path: string]: BehaviorSubject<any> } = { }
|
private watchedNodes: { [path: string]: ReplaySubject<any> } = {}
|
||||||
private stash = new BTree<number, StashEntry>()
|
private stash = new BTree<number, StashEntry>()
|
||||||
|
|
||||||
constructor (
|
constructor(
|
||||||
private readonly http: Http<T>,
|
private readonly http: Http<T>,
|
||||||
private readonly initialCache: DBCache<T>,
|
private readonly initialCache: DBCache<T>,
|
||||||
) {
|
) {
|
||||||
@@ -22,24 +22,25 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
this.sequence$ = new BehaviorSubject(initialCache.sequence)
|
this.sequence$ = new BehaviorSubject(initialCache.sequence)
|
||||||
}
|
}
|
||||||
|
|
||||||
watch$ (): Observable<T>
|
watch$(): Observable<T>
|
||||||
watch$<P1 extends keyof T> (p1: P1): Observable<T[P1]>
|
watch$<P1 extends keyof T>(p1: P1): Observable<NonNullable<T[P1]>>
|
||||||
watch$<P1 extends keyof T, P2 extends keyof T[P1]> (p1: P1, p2: P2): Observable<T[P1][P2]>
|
watch$<P1 extends keyof T, P2 extends keyof NonNullable<T[P1]>>(p1: P1, p2: P2): Observable<NonNullable<NonNullable<T[P1]>[P2]>>
|
||||||
watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2]> (p1: P1, p2: P2, p3: P3): Observable<T[P1][P2][P3]>
|
watch$<P1 extends keyof T, P2 extends keyof NonNullable<T[P1]>, P3 extends keyof NonNullable<NonNullable<T[P1]>[P2]>>(p1: P1, p2: P2, p3: P3): Observable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>>
|
||||||
watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3]> (p1: P1, p2: P2, p3: P3, p4: P4): Observable<T[P1][P2][P3][P4]>
|
watch$<P1 extends keyof T, P2 extends keyof NonNullable<T[P1]>, P3 extends keyof NonNullable<NonNullable<T[P1]>[P2]>, P4 extends keyof NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>>(p1: P1, p2: P2, p3: P3, p4: P4): Observable<NonNullable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>[P4]>>
|
||||||
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 NonNullable<T[P1]>, P3 extends keyof NonNullable<NonNullable<T[P1]>[P2]>, P4 extends keyof NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>, P5 extends keyof NonNullable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>[P4]>>(p1: P1, p2: P2, p3: P3, p4: P4, p5: P5): Observable<NonNullable<NonNullable<NonNullable<NonNullable<NonNullable<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 NonNullable<T[P1]>, P3 extends keyof NonNullable<NonNullable<T[P1]>[P2]>, P4 extends keyof NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>, P5 extends keyof NonNullable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>[P4]>, P6 extends keyof NonNullable<NonNullable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>[P4]>[P5]>>(p1: P1, p2: P2, p3: P3, p4: P4, p5: P5, p6: P6): Observable<NonNullable<NonNullable<NonNullable<NonNullable<NonNullable<NonNullable<T[P1]>[P2]>[P3]>[P4]>[P5]>[P6]>>
|
||||||
watch$ (...args: (string | number)[]): Observable<any> {
|
watch$(...args: (string | number)[]): Observable<any> {
|
||||||
const path = `/${args.join('/')}`
|
const path = `/${args.join('/')}`
|
||||||
if (!this.watchedNodes[path]) {
|
if (!this.watchedNodes[path]) {
|
||||||
this.watchedNodes[path] = new BehaviorSubject(getValueByPointer(this.cache.data, path))
|
this.watchedNodes[path] = new ReplaySubject(1)
|
||||||
|
this.updateValue(path)
|
||||||
}
|
}
|
||||||
return this.watchedNodes[path].asObservable()
|
return this.watchedNodes[path].asObservable()
|
||||||
}
|
}
|
||||||
|
|
||||||
update (update: Update<T>): void {
|
update(update: Update<T>): void {
|
||||||
if (this.isRevision(update)) {
|
if (this.isRevision(update)) {
|
||||||
// if old or known, return
|
// if old or known, return
|
||||||
if (update.id <= this.cache.sequence || this.stash.get(update.id)) return
|
if (update.id <= this.cache.sequence || this.stash.get(update.id)) return
|
||||||
this.handleRevision(update)
|
this.handleRevision(update)
|
||||||
} else {
|
} else {
|
||||||
@@ -47,18 +48,24 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
reset (): void {
|
reset(): void {
|
||||||
Object.values(this.watchedNodes).forEach(node => node.complete())
|
Object.values(this.watchedNodes).forEach(node => node.complete())
|
||||||
this.watchedNodes = { }
|
this.watchedNodes = {}
|
||||||
this.stash.clear()
|
this.stash.clear()
|
||||||
this.sequence$.next(0)
|
this.sequence$.next(0)
|
||||||
this.cache = {
|
this.cache = {
|
||||||
sequence: 0,
|
sequence: 0,
|
||||||
data: { } as any,
|
data: {} as any,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private handleRevision (revision: Revision): void {
|
private updateValue(path: string): void {
|
||||||
|
const value = getValueByPointer(this.cache.data, path)
|
||||||
|
|
||||||
|
this.watchedNodes[path].next(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
private handleRevision(revision: Revision): void {
|
||||||
// stash the revision
|
// stash the revision
|
||||||
this.stash.set(revision.id, { revision, undo: [] })
|
this.stash.set(revision.id, { revision, undo: [] })
|
||||||
|
|
||||||
@@ -70,7 +77,7 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
this.processStashed(revision.id)
|
this.processStashed(revision.id)
|
||||||
}
|
}
|
||||||
|
|
||||||
private handleDump ({ value, id }: Dump<T>): void {
|
private handleDump({ value, id }: Dump<T>): void {
|
||||||
this.cache.data = { ...value }
|
this.cache.data = { ...value }
|
||||||
this.stash.deleteRange(this.cache.sequence, id, false)
|
this.stash.deleteRange(this.cache.sequence, id, false)
|
||||||
this.updateWatchedNodes('')
|
this.updateWatchedNodes('')
|
||||||
@@ -78,12 +85,12 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
this.processStashed(id + 1)
|
this.processStashed(id + 1)
|
||||||
}
|
}
|
||||||
|
|
||||||
private processStashed (id: number): void {
|
private processStashed(id: number): void {
|
||||||
this.undoRevisions(id)
|
this.undoRevisions(id)
|
||||||
this.applyRevisions(id)
|
this.applyRevisions(id)
|
||||||
}
|
}
|
||||||
|
|
||||||
private undoRevisions (id: number): void {
|
private undoRevisions(id: number): void {
|
||||||
let stashEntry = this.stash.get(this.stash.maxKey() as number)
|
let stashEntry = this.stash.get(this.stash.maxKey() as number)
|
||||||
|
|
||||||
while (stashEntry && stashEntry.revision.id > id) {
|
while (stashEntry && stashEntry.revision.id > id) {
|
||||||
@@ -92,7 +99,7 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private applyRevisions (id: number): void {
|
private applyRevisions(id: number): void {
|
||||||
let revision = this.stash.get(id)?.revision
|
let revision = this.stash.get(id)?.revision
|
||||||
while (revision) {
|
while (revision) {
|
||||||
let undo: Operation<unknown>[] = []
|
let undo: Operation<unknown>[] = []
|
||||||
@@ -129,7 +136,7 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
this.stash.deleteRange(0, this.cache.sequence, false)
|
this.stash.deleteRange(0, this.cache.sequence, false)
|
||||||
}
|
}
|
||||||
|
|
||||||
private updateWatchedNodes (revisionPath: string) {
|
private updateWatchedNodes(revisionPath: string) {
|
||||||
const kill = (path: string) => {
|
const kill = (path: string) => {
|
||||||
this.watchedNodes[path].complete()
|
this.watchedNodes[path].complete()
|
||||||
delete this.watchedNodes[path]
|
delete this.watchedNodes[path]
|
||||||
@@ -139,18 +146,17 @@ export class Store<T extends { [key: string]: any }> {
|
|||||||
if (this.watchedNodes[path].observers.length === 0) return kill(path)
|
if (this.watchedNodes[path].observers.length === 0) return kill(path)
|
||||||
|
|
||||||
if (path.includes(revisionPath) || revisionPath.includes(path)) {
|
if (path.includes(revisionPath) || revisionPath.includes(path)) {
|
||||||
const val = getValueByPointer(this.cache.data, path)
|
this.updateValue(path)
|
||||||
this.watchedNodes[path].next(val)
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
private updateSequence (sequence: number): void {
|
private updateSequence(sequence: number): void {
|
||||||
this.cache.sequence = sequence
|
this.cache.sequence = sequence
|
||||||
this.sequence$.next(sequence)
|
this.sequence$.next(sequence)
|
||||||
}
|
}
|
||||||
|
|
||||||
private isRevision (update: Update<T>): update is Revision {
|
private isRevision(update: Update<T>): update is Revision {
|
||||||
return 'patch' in update
|
return 'patch' in update
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user