mirror of
https://github.com/Start9Labs/patch-db.git
synced 2026-03-26 02:11:54 +00:00
Merge branch 'master' of https://github.com/Start9Labs/patch-db
This commit is contained in:
@@ -1,4 +1,4 @@
|
|||||||
import { DBCache, PatchOp } from './types'
|
import { Dump, PatchOp } from './types'
|
||||||
|
|
||||||
export interface BaseOperation {
|
export interface BaseOperation {
|
||||||
path: string
|
path: string
|
||||||
@@ -37,10 +37,10 @@ export function getValueByPointer<T extends Record<string, T>>(
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function applyOperation<T>(
|
export function applyOperation<T>(
|
||||||
doc: DBCache<Record<string, any>>,
|
doc: Dump<Record<string, any>>,
|
||||||
{ path, op, value }: Operation<T> & { value?: T },
|
{ path, op, value }: Operation<T> & { value?: T },
|
||||||
) {
|
) {
|
||||||
doc.data = recursiveApply(doc.data, arrayFromPath(path), op, value)
|
doc.value = recursiveApply(doc.value, arrayFromPath(path), op, value)
|
||||||
}
|
}
|
||||||
|
|
||||||
export function arrayFromPath(path: string): string[] {
|
export function arrayFromPath(path: string): string[] {
|
||||||
@@ -61,7 +61,7 @@ export function pathFromArray(args: Array<string | number>): string {
|
|||||||
args
|
args
|
||||||
.map(a =>
|
.map(a =>
|
||||||
String(a)
|
String(a)
|
||||||
// do not change order, "~" needs to be replaced first
|
// do not change order, "~" needs to be replaced first
|
||||||
.replace(new RegExp('~', 'g'), '~0')
|
.replace(new RegExp('~', 'g'), '~0')
|
||||||
.replace(new RegExp('/', 'g'), '~1'),
|
.replace(new RegExp('/', 'g'), '~1'),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { Bootstrapper, DBCache, Dump, Revision, Update } from './types'
|
import { Dump, Revision, Update } from './types'
|
||||||
import {
|
import {
|
||||||
BehaviorSubject,
|
BehaviorSubject,
|
||||||
filter,
|
filter,
|
||||||
@@ -24,24 +24,21 @@ export class PatchDB<T extends { [key: string]: any }> {
|
|||||||
}
|
}
|
||||||
} = {}
|
} = {}
|
||||||
|
|
||||||
readonly cache$ = new BehaviorSubject<DBCache<T>>({
|
constructor(
|
||||||
sequence: 0,
|
private readonly source$: Observable<Update<T>[]>,
|
||||||
data: {} as T,
|
private readonly cache$ = new BehaviorSubject<Dump<T>>({
|
||||||
})
|
id: 0,
|
||||||
|
value: {} as T,
|
||||||
|
}),
|
||||||
|
) {}
|
||||||
|
|
||||||
constructor(private readonly source$: Observable<Update<T>[]>) {}
|
start() {
|
||||||
|
|
||||||
start(bootstrapper: Bootstrapper<T>) {
|
|
||||||
if (this.sub) return
|
if (this.sub) return
|
||||||
|
|
||||||
const initialCache = bootstrapper.init()
|
|
||||||
this.cache$.next(initialCache)
|
|
||||||
|
|
||||||
this.sub = this.source$
|
this.sub = this.source$
|
||||||
.pipe(withLatestFrom(this.cache$))
|
.pipe(withLatestFrom(this.cache$))
|
||||||
.subscribe(([updates, cache]) => {
|
.subscribe(([updates, cache]) => {
|
||||||
this.proccessUpdates(updates, cache)
|
this.proccessUpdates(updates, cache)
|
||||||
bootstrapper.update(cache)
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -52,7 +49,7 @@ export class PatchDB<T extends { [key: string]: any }> {
|
|||||||
this.watchedNodes = {}
|
this.watchedNodes = {}
|
||||||
this.sub.unsubscribe()
|
this.sub.unsubscribe()
|
||||||
this.sub = null
|
this.sub = null
|
||||||
this.cache$.next({ sequence: 0, data: {} as T })
|
this.cache$.next({ id: 0, value: {} as T })
|
||||||
}
|
}
|
||||||
|
|
||||||
watch$(): Observable<T>
|
watch$(): Observable<T>
|
||||||
@@ -131,14 +128,13 @@ export class PatchDB<T extends { [key: string]: any }> {
|
|||||||
>
|
>
|
||||||
watch$(...args: (string | number)[]): Observable<any> {
|
watch$(...args: (string | number)[]): Observable<any> {
|
||||||
return this.cache$.pipe(
|
return this.cache$.pipe(
|
||||||
filter(({ sequence }) => !!sequence),
|
filter(({ id }) => !!id),
|
||||||
take(1),
|
take(1),
|
||||||
switchMap(({ data }) => {
|
switchMap(({ value }) => {
|
||||||
const path = pathFromArray(args)
|
const path = pathFromArray(args)
|
||||||
if (!this.watchedNodes[path]) {
|
if (!this.watchedNodes[path]) {
|
||||||
const value = getValueByPointer(data, path)
|
|
||||||
this.watchedNodes[path] = {
|
this.watchedNodes[path] = {
|
||||||
subject: new BehaviorSubject(value),
|
subject: new BehaviorSubject(getValueByPointer(value, path)),
|
||||||
pathArr: arrayFromPath(path),
|
pathArr: arrayFromPath(path),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -147,21 +143,21 @@ export class PatchDB<T extends { [key: string]: any }> {
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
proccessUpdates(updates: Update<T>[], cache: DBCache<T>) {
|
proccessUpdates(updates: Update<T>[], cache: Dump<T>) {
|
||||||
updates.forEach(update => {
|
updates.forEach(update => {
|
||||||
if (this.isRevision(update)) {
|
if (this.isRevision(update)) {
|
||||||
const expected = cache.sequence + 1
|
const expected = cache.id + 1
|
||||||
if (update.id < expected) return
|
if (update.id < expected) return
|
||||||
this.handleRevision(update, cache)
|
this.handleRevision(update, cache)
|
||||||
} else {
|
} else {
|
||||||
this.handleDump(update, cache)
|
this.handleDump(update, cache)
|
||||||
}
|
}
|
||||||
cache.sequence = update.id
|
cache.id = update.id
|
||||||
})
|
})
|
||||||
this.cache$.next(cache)
|
this.cache$.next(cache)
|
||||||
}
|
}
|
||||||
|
|
||||||
private handleRevision(revision: Revision, cache: DBCache<T>): void {
|
private handleRevision(revision: Revision, cache: Dump<T>): void {
|
||||||
// apply opperations
|
// apply opperations
|
||||||
revision.patch.forEach(op => {
|
revision.patch.forEach(op => {
|
||||||
applyOperation(cache, op)
|
applyOperation(cache, op)
|
||||||
@@ -172,14 +168,14 @@ export class PatchDB<T extends { [key: string]: any }> {
|
|||||||
const arr = arrayFromPath(path)
|
const arr = arrayFromPath(path)
|
||||||
return startsWith(pathArr, arr) || startsWith(arr, pathArr)
|
return startsWith(pathArr, arr) || startsWith(arr, pathArr)
|
||||||
})
|
})
|
||||||
if (match) this.updateWatchedNode(watchedPath, cache.data)
|
if (match) this.updateWatchedNode(watchedPath, cache.value)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
private handleDump(dump: Dump<T>, cache: DBCache<T>): void {
|
private handleDump(dump: Dump<T>, cache: Dump<T>): void {
|
||||||
cache.data = { ...dump.value }
|
cache.value = { ...dump.value }
|
||||||
Object.keys(this.watchedNodes).forEach(watchedPath => {
|
Object.keys(this.watchedNodes).forEach(watchedPath => {
|
||||||
this.updateWatchedNode(watchedPath, cache.data)
|
this.updateWatchedNode(watchedPath, cache.value)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,10 @@
|
|||||||
import { Operation } from './json-patch-lib'
|
import { Operation } from './json-patch-lib'
|
||||||
|
|
||||||
// revise a collection of nodes.
|
|
||||||
export type Revision = {
|
export type Revision = {
|
||||||
id: number
|
id: number
|
||||||
patch: Operation<unknown>[]
|
patch: Operation<unknown>[]
|
||||||
}
|
}
|
||||||
|
|
||||||
// dump/replace the entire store with T
|
|
||||||
export type Dump<T> = { id: number; value: T }
|
export type Dump<T> = { id: number; value: T }
|
||||||
|
|
||||||
export type Update<T> = Revision | Dump<T>
|
export type Update<T> = Revision | Dump<T>
|
||||||
@@ -16,13 +14,3 @@ export enum PatchOp {
|
|||||||
REMOVE = 'remove',
|
REMOVE = 'remove',
|
||||||
REPLACE = 'replace',
|
REPLACE = 'replace',
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface Bootstrapper<T extends Record<string, any>> {
|
|
||||||
init(): DBCache<T>
|
|
||||||
update(cache: DBCache<T>): void
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface DBCache<T extends Record<string, any>> {
|
|
||||||
sequence: number
|
|
||||||
data: T
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user