start

client/lib/patch-db.ts

@@ -1,4 +1,4 @@
-import { merge, Observable } from 'rxjs'
+import { merge, Observable, of } from 'rxjs'
 import { concatMap, finalize, tap } from 'rxjs/operators'
 import { Source } from './source/source'
 import { Store } from './store'
@@ -6,21 +6,21 @@ import { DBCache } from './types'
 
 export { Operation } from 'fast-json-patch'
 
-export class PatchDB<T extends object> {
+export class PatchDB<T> {
   store: Store<T>
 
   constructor (
     private readonly sources: Source<T>[],
-    readonly cache: DBCache<T>,
+    private readonly initialCache: DBCache<T>,
   ) {
-    this.store = new Store(cache)
+    this.store = new Store(this.initialCache)
   }
 
   sync$ (): Observable<DBCache<T>> {
     return merge(...this.sources.map(s => s.watch$(this.store)))
     .pipe(
       tap(update => this.store.update(update)),
-      concatMap(() => this.store.watchCache$()),
+      concatMap(() => of(this.store.cache)),
       finalize(() => {
         this.store.reset()
       }),

client/lib/source/poll-source.ts

@@ -1,5 +1,5 @@
 import { BehaviorSubject, concat, from, Observable, of } from 'rxjs'
-import { concatMap, delay, map, skip, switchMap, take, tap } from 'rxjs/operators'
+import { concatMap, delay, skip, switchMap, take, tap } from 'rxjs/operators'
 import { Store } from '../store'
 import { Http, Update } from '../types'
 import { Source } from './source'
@@ -16,17 +16,13 @@ export class PollSource<T> implements Source<T> {
   ) { }
 
   watch$ (store: Store<T>): Observable<Update<T>> {
-    const sequence$ = store.watchCache$()
-    .pipe(
-      map(cache => cache.sequence),
-    )
-
     const polling$ = new BehaviorSubject('')
 
-    const updates$ = of('').pipe(
+    const updates$ = of({ })
-      concatMap(_ => sequence$),
+    .pipe(
-      take(1),
+      concatMap(_ => store.sequence$),
       concatMap(seq => this.http.getRevisions(seq)),
+      take(1),
     )
 
     const delay$ = of([]).pipe(

client/lib/store.ts

@@ -1,20 +1,17 @@
-import { BehaviorSubject, from, Observable } from 'rxjs'
-import { observable } from 'mobx'
-import { toStream } from 'mobx-utils'
 import { DBCache, Dump, Revision, Update } from './types'
 import { applyPatch } from 'fast-json-patch'
+import { BehaviorSubject, from, Observable } from 'rxjs'
+import { toStream } from 'mobx-utils'
 
 export class Store<T> {
-  sequence: number
+  cache: DBCache<T>
-  o: { data: T | { } }
+  sequence$: BehaviorSubject<number>
-  cache$: BehaviorSubject<DBCache<T>>
 
   constructor (
     readonly initialCache: DBCache<T>,
   ) {
-    this.sequence = initialCache.sequence
+    this.cache = initialCache
-    this.o = observable({ data: this.initialCache.data })
+    this.sequence$ = new BehaviorSubject(initialCache.sequence)
-    this.cache$ = new BehaviorSubject(initialCache)
   }
 
   watch$ (): Observable<T>
@@ -28,33 +25,25 @@ export class Store<T> {
     return from(toStream(() => this.peekNode(...args), true))
   }
 
-  watchCache$ (): Observable<DBCache<T>> {
-    return this.cache$.asObservable()
-  }
-
   update (update: Update<T>): void {
     if ((update as Revision).patch) {
-      if (this.sequence + 1 !== update.id) throw new Error(`Outdated sequence: current: ${this.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.o.data, (update as Revision).patch, true, true)
+      applyPatch(this.cache.data, (update as Revision).patch, true, true)
     } else {
-      this.o.data = (update as Dump<T>).value
+      this.cache.data = (update as Dump<T>).value
     }
-
+    this.cache.sequence = update.id
-    this.sequence = update.id
+    this.sequence$.next(this.cache.sequence)
-
-    this.cache$.next({ sequence: this.sequence, data: this.o.data })
   }
 
   reset (): void {
-    this.cache$.next({
+    this.cache.sequence = 0
-      sequence: 0,
+    this.cache.data = { } as T
-      data: { },
-    })
   }
 
   private peekNode (...args: (string | number)[]): any {
     try {
-      return args.reduce((acc, next) => (acc as any)[`${next}`], this.o.data)
+      return args.reduce((acc, next) => (acc as any)[`${next}`], this.cache.data)
     } catch (e) {
       return undefined
     }

client/lib/types.ts

@@ -25,5 +25,5 @@ export interface Bootstrapper<T> {
 
 export interface DBCache<T>{
   sequence: number,
-  data: T | { }
+  data: T
 }