remove sources entirely

client/index.ts

@@ -1,6 +1,3 @@
-export * from './lib/source/mock-source'
-export * from './lib/source/ws-source'
-export * from './lib/source/source'
 export * from './lib/json-patch-lib'
 export * from './lib/patch-db'
 export * from './lib/store'

client/lib/patch-db.ts

@@ -1,28 +1,18 @@
-import { map, shareReplay, Subject, timer } from 'rxjs'
+import { map, Observable, shareReplay } from 'rxjs'
-import { catchError, concatMap, tap } from 'rxjs/operators'
+import { tap } from 'rxjs/operators'
 import { Store } from './store'
-import { DBCache, Http } from './types'
+import { DBCache, Update } from './types'
-import { Source } from './source/source'
 
 export class PatchDB<T> {
-  public store: Store<T> = new Store(this.http, this.initialCache)
+  public store: Store<T> = new Store(this.initialCache)
-  public connectionError$ = new Subject<any>()
+  public cache$ = this.source$.pipe(
-  public cache$ = this.source.watch$(this.store).pipe(
+    tap(res => this.store.update(res)),
-    catchError((e, watch$) => {
-      this.connectionError$.next(e)
-      return timer(4000).pipe(concatMap(() => watch$))
-    }),
-    tap(res => {
-      this.connectionError$.next(null)
-      this.store.update(res)
-    }),
     map(_ => this.store.cache),
     shareReplay(1),
   )
 
   constructor(
-    private readonly source: Source<T>,
+    private readonly source$: Observable<Update<T>>,
-    private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {}
 }

client/lib/source/mock-source.ts

@@ -1,11 +0,0 @@
-import { Observable } from 'rxjs'
-import { Update } from '../types'
-import { Source } from './source'
-
-export class MockSource<T> implements Source<T> {
-  constructor(private readonly seed: Observable<Update<T>>) {}
-
-  watch$(): Observable<Update<T>> {
-    return this.seed
-  }
-}

client/lib/source/source.ts

@@ -1,7 +0,0 @@
-import { Observable } from 'rxjs'
-import { Store } from '../store'
-import { Update } from '../types'
-
-export interface Source<T> {
-  watch$(store?: Store<T>): Observable<Update<T>>
-}

client/lib/source/ws-source.ts

@@ -1,12 +0,0 @@
-import { Observable, timeout } from 'rxjs'
-import { webSocket } from 'rxjs/webSocket'
-import { Update } from '../types'
-import { Source } from './source'
-
-export class WebsocketSource<T> implements Source<T> {
-  constructor(private readonly url: string) {}
-
-  watch$(): Observable<Update<T>> {
-    return webSocket<Update<T>>(this.url).pipe(timeout({ first: 21000 }))
-  }
-}

client/lib/store.ts

@@ -1,4 +1,4 @@
-import { DBCache, Dump, Http, Revision, Update } from './types'
+import { DBCache, Dump, Revision, Update } from './types'
 import { BehaviorSubject, Observable, ReplaySubject } from 'rxjs'
 import { applyOperation, getValueByPointer, Operation } from './json-patch-lib'
 import BTree from 'sorted-btree'
@@ -13,7 +13,7 @@ export class Store<T extends { [key: string]: any }> {
   private watchedNodes: { [path: string]: ReplaySubject<any> } = {}
   private stash = new BTree<number, StashEntry>()
 
-  constructor(private readonly http: Http<T>, public cache: DBCache<T>) {}
+  constructor(public cache: DBCache<T>) {}
 
   watch$(): Observable<T>
   watch$<P1 extends keyof T>(p1: P1): Observable<NonNullable<T[P1]>>
@@ -127,14 +127,7 @@ export class Store<T extends { [key: string]: any }> {
   }
 
   private handleRevision(revision: Revision): void {
-    // stash the revision
     this.stash.set(revision.id, { revision, undo: [] })
-
-    // if revision is futuristic, fetch missing revisions
-    if (revision.id > this.cache.sequence + 1) {
-      this.http.getRevisions(this.cache.sequence)
-    }
-
     this.processStashed(revision.id)
   }
 

client/lib/types.ts

@@ -18,11 +18,6 @@ export enum PatchOp {
   REPLACE = 'replace',
 }
 
-export interface Http<T> {
-  getRevisions(since: number): Promise<Revision[] | Dump<T>>
-  getDump(): Promise<Dump<T>>
-}
-
 export interface Bootstrapper<T> {
   init(): Promise<DBCache<T>>
   update(cache: DBCache<T>): Promise<void>