Refactor sources (#40)

* major refactor to exclude multiple sources

Co-authored-by: Matt Hill <matthewonthemoon@gmail.com>

client/lib/patch-db.ts

@@ -1,44 +1,18 @@
-import { EMPTY, merge, Observable, ReplaySubject, Subject } from 'rxjs'
-import { catchError, switchMap } from 'rxjs/operators'
+import { map, Observable, shareReplay } from 'rxjs'
+import { tap } from 'rxjs/operators'
 import { Store } from './store'
-import { DBCache, Http } from './types'
-import { RPCError } from './source/ws-source'
-import { Source } from './source/source'
+import { DBCache, Update } from './types'
 
 export class PatchDB<T> {
-  public store: Store<T> = new Store(this.http, this.initialCache)
-  public connectionError$ = new Subject<Error>()
-  public rpcError$ = new Subject<RPCError>()
-  public cache$ = new ReplaySubject<DBCache<T>>(1)
-
-  private sub = this.sources$
-    .pipe(
-      switchMap(sources =>
-        merge(...sources.map(s => s.watch$(this.store))).pipe(
-          catchError(e => {
-            this.connectionError$.next(e)
-
-            return EMPTY
-          }),
-        ),
-      ),
-    )
-    .subscribe(res => {
-      if ('result' in res) {
-        this.store.update(res.result)
-        this.cache$.next(this.store.cache)
-      } else {
-        this.rpcError$.next(res)
-      }
-    })
+  public store: Store<T> = new Store(this.initialCache)
+  public cache$ = this.source$.pipe(
+    tap(res => this.store.update(res)),
+    map(_ => this.store.cache),
+    shareReplay(1),
+  )
 
   constructor(
-    private readonly sources$: Observable<Source<T>[]>,
-    private readonly http: Http<T>,
+    private readonly source$: Observable<Update<T>>,
     private readonly initialCache: DBCache<T>,
   ) {}
-
-  clean() {
-    this.sub.unsubscribe()
-  }
 }

client/lib/source/mock-source.ts

@@ -1,13 +0,0 @@
-import { Observable } from 'rxjs'
-import { map } from 'rxjs/operators'
-import { Update } from '../types'
-import { Source } from './source'
-import { RPCResponse } from './ws-source'
-
-export class MockSource<T> implements Source<T> {
-  constructor(private readonly seed: Observable<Update<T>>) {}
-
-  watch$(): Observable<RPCResponse<Update<T>>> {
-    return this.seed.pipe(map(result => ({ result, jsonrpc: '2.0' })))
-  }
-}

client/lib/source/poll-source.ts

@@ -1,55 +0,0 @@
-import { BehaviorSubject, concat, from, Observable, of } from 'rxjs'
-import {
-  concatMap,
-  delay,
-  map,
-  skip,
-  switchMap,
-  take,
-  tap,
-} from 'rxjs/operators'
-import { Store } from '../store'
-import { Http, Update } from '../types'
-import { Source } from './source'
-import { RPCResponse } from './ws-source'
-
-export type PollConfig = {
-  cooldown: number
-}
-
-export class PollSource<T> implements Source<T> {
-  constructor(
-    private readonly pollConfig: PollConfig,
-    private readonly http: Http<T>,
-  ) {}
-
-  watch$(store: Store<T>): Observable<RPCResponse<Update<T>>> {
-    const polling$ = new BehaviorSubject('')
-
-    const updates$ = of({}).pipe(
-      concatMap(_ => store.sequence$),
-      concatMap(seq => this.http.getRevisions(seq)),
-      take(1),
-    )
-
-    const delay$ = of([]).pipe(
-      delay(this.pollConfig.cooldown),
-      tap(_ => polling$.next('')),
-      skip(1),
-    )
-
-    const poll$ = concat(updates$, delay$)
-
-    return polling$.pipe(
-      switchMap(_ => poll$),
-      concatMap(res => {
-        if (Array.isArray(res)) {
-          return from(res) // takes Revision[] and converts it into Observable<Revision>
-        } else {
-          return of(res) // takes Dump<T> and converts it into Observable<Dump<T>>
-        }
-      }),
-      map(result => ({ result, jsonrpc: '2.0' })),
-    )
-  }
-}

client/lib/source/source.ts

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

client/lib/source/ws-source.ts

@@ -1,58 +0,0 @@
-import { Observable } from 'rxjs'
-import {
-  webSocket,
-  WebSocketSubject,
-  WebSocketSubjectConfig,
-} from 'rxjs/webSocket'
-import { Update } from '../types'
-import { Source } from './source'
-
-export class WebsocketSource<T> implements Source<T> {
-  private websocket$: WebSocketSubject<RPCResponse<Update<T>>> | undefined
-
-  constructor(private readonly url: string) {}
-
-  watch$(): Observable<RPCResponse<Update<T>>> {
-    const fullConfig: WebSocketSubjectConfig<RPCResponse<Update<T>>> = {
-      url: this.url,
-      openObserver: {
-        next: () => {
-          this.websocket$!.next(document.cookie as any)
-        },
-      },
-    }
-    this.websocket$ = webSocket(fullConfig)
-    return this.websocket$
-  }
-}
-
-interface RPCBase {
-  jsonrpc: '2.0'
-}
-
-export interface RPCSuccess<T> extends RPCBase {
-  result: T
-}
-
-export interface RPCError extends RPCBase {
-  error: {
-    code: number // 34 means unauthenticated
-    message: string
-    data: {
-      details: string
-    }
-  }
-}
-export type RPCResponse<T> = RPCSuccess<T> | RPCError
-
-class RpcError {
-  code: number
-  message: string
-  details: string
-
-  constructor(e: RPCError['error']) {
-    this.code = e.code
-    this.message = e.message
-    this.details = e.data.details
-  }
-}

client/lib/store.ts

@@ -1,5 +1,5 @@
-import { DBCache, Dump, Http, Revision, Update } from './types'
-import { BehaviorSubject, Observable, ReplaySubject } from 'rxjs'
+import { DBCache, Dump, Revision, Update } from './types'
+import { BehaviorSubject, Observable } from 'rxjs'
 import { applyOperation, getValueByPointer, Operation } from './json-patch-lib'
 import BTree from 'sorted-btree'
 
@@ -9,18 +9,11 @@ export interface StashEntry {
 }
 
 export class Store<T extends { [key: string]: any }> {
-  cache: DBCache<T>
-  sequence$: BehaviorSubject<number>
-  private watchedNodes: { [path: string]: ReplaySubject<any> } = {}
+  readonly sequence$ = new BehaviorSubject(this.cache.sequence)
+  private watchedNodes: { [path: string]: BehaviorSubject<any> } = {}
   private stash = new BTree<number, StashEntry>()
 
-  constructor(
-    private readonly http: Http<T>,
-    private readonly initialCache: DBCache<T>,
-  ) {
-    this.cache = this.initialCache
-    this.sequence$ = new BehaviorSubject(initialCache.sequence)
-  }
+  constructor(public cache: DBCache<T>) {}
 
   watch$(): Observable<T>
   watch$<P1 extends keyof T>(p1: P1): Observable<NonNullable<T[P1]>>
@@ -98,18 +91,32 @@ export class Store<T extends { [key: string]: any }> {
   >
   watch$(...args: (string | number)[]): Observable<any> {
     const path = `/${args.join('/')}`
-    if (!this.watchedNodes[path]) {
-      this.watchedNodes[path] = new ReplaySubject(1)
+
+    return new Observable(subscriber => {
+      const value = getValueByPointer(this.cache.data, path)
+      const source = this.watchedNodes[path] || new BehaviorSubject(value)
+      const subscription = source.subscribe(subscriber)
+
+      this.watchedNodes[path] = source
       this.updateValue(path)
-    }
-    return this.watchedNodes[path].asObservable()
+
+      return () => {
+        subscription.unsubscribe()
+
+        if (!source.observed) {
+          source.complete()
+          delete this.watchedNodes[path]
+        }
+      }
+    })
   }
 
   update(update: Update<T>): void {
     if (this.isRevision(update)) {
-      // if old or known, return
-      if (update.id <= this.cache.sequence || this.stash.get(update.id)) return
-      this.handleRevision(update)
+      // Handle revision if new and not known
+      if (update.id > this.cache.sequence && !this.stash.get(update.id)) {
+        this.handleRevision(update)
+      }
     } else {
       this.handleDump(update)
     }
@@ -133,14 +140,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)
   }
 
@@ -204,14 +204,7 @@ export class Store<T extends { [key: string]: any }> {
   }
 
   private updateWatchedNodes(revisionPath: string) {
-    const kill = (path: string) => {
-      this.watchedNodes[path].complete()
-      delete this.watchedNodes[path]
-    }
-
     Object.keys(this.watchedNodes).forEach(path => {
-      if (this.watchedNodes[path].observers.length === 0) return kill(path)
-
       if (path.includes(revisionPath) || revisionPath.includes(path)) {
         this.updateValue(path)
       }

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>