remove connection status from lib

client/lib/patch-db.ts

@@ -1,32 +1,27 @@
-import { Observable } from 'rxjs'
+import { merge, Observable } from 'rxjs'
-import { concatMap, finalize, map, tap } from 'rxjs/operators'
+import { concatMap, finalize, tap } from 'rxjs/operators'
 import { Source } from './source/source'
 import { Store } from './store'
 import { DBCache } from './types'
+
 export { Operation } from 'fast-json-patch'
 
 export class PatchDB<T extends object> {
   store: Store<T>
-  connectionStatus$ = this.source.connectionStatus$
 
   constructor (
-    private readonly source: Source<T>,
+    private readonly sources: Source<T>[],
     readonly cache: DBCache<T>,
   ) {
     this.store = new Store(cache)
   }
 
   sync$ (): Observable<DBCache<T>> {
-    console.log('PATCHDB - sync$()')
+    return merge(...this.sources.map(s => s.watch$(this.store)))
-
+    .pipe(
-    const sequence$ = this.store.watchAll$().pipe(map(cache => cache.sequence))
+      tap(update => this.store.update(update)),
-    // nested concatMaps, as it is written, ensure sync is not run for update2 until handleSyncResult is complete for update1.
+      concatMap(() => this.store.watchCache$()),
-    // flat concatMaps would allow many syncs to run while handleSyncResult was hanging. We can consider such an idea if performance requires it.
-    return this.source.watch$(sequence$).pipe(
-      tap(update => console.log('PATCHDB - source updated:', update)),
-      concatMap(update => this.store.update$(update)),
       finalize(() => {
-        console.log('PATCHDB - FINALIZING sync$()')
         this.store.reset()
       }),
     )

client/lib/source/poll-source.ts

@@ -1,6 +1,7 @@
 import { BehaviorSubject, concat, from, Observable, of } from 'rxjs'
-import { catchError, concatMap, delay, skip, switchMap, take, tap } from 'rxjs/operators'
+import { concatMap, delay, map, skip, switchMap, take, tap } from 'rxjs/operators'
-import { ConnectionStatus, Http, Update } from '../types'
+import { Store } from '../store'
+import { Http, Update } from '../types'
 import { Source } from './source'
 
 export type PollConfig = {
@@ -8,26 +9,24 @@ export type PollConfig = {
 }
 
 export class PollSource<T> implements Source<T> {
-  connectionStatus$ = new BehaviorSubject(ConnectionStatus.Initializing)
 
   constructor (
     private readonly pollConfig: PollConfig,
     private readonly http: Http<T>,
   ) { }
 
-  watch$ (sequence$: Observable<number>): Observable<Update<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(
       concatMap(_ => sequence$),
       take(1),
       concatMap(seq => this.http.getRevisions(seq)),
-      tap(_ => this.connectionStatus$.next(ConnectionStatus.Connected)),
-      catchError(e => {
-        console.error(e)
-        this.connectionStatus$.next(ConnectionStatus.Disconnected)
-        return of([])
-      }),
     )
 
     const delay$ = of([]).pipe(

client/lib/source/source.ts

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

client/lib/source/ws-source.ts

@@ -1,16 +1,14 @@
-import { BehaviorSubject, Observable } from 'rxjs'
+import { Observable } from 'rxjs'
 import { webSocket, WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket'
-import { ConnectionStatus, Update } from '../types'
+import { Update } from '../types'
 import { Source } from './source'
 
 export class WebsocketSource<T> implements Source<T> {
-  connectionStatus$ = new BehaviorSubject(ConnectionStatus.Initializing)
   private websocket$: WebSocketSubject<Update<T>> | undefined
 
   constructor (
     private readonly url: string,
-  ) {
+  ) { }
-  }
 
   watch$ (): Observable<Update<T>> {
     const fullConfig: WebSocketSubjectConfig<Update<T>> = {
@@ -18,21 +16,9 @@ export class WebsocketSource<T> implements Source<T> {
       openObserver: {
         next: () => {
           console.log('WebSocket connection open')
-          this.connectionStatus$.next(ConnectionStatus.Connected)
           this.websocket$!.next('open message' as any)
         },
       },
-      closeObserver: {
-        next: () => {
-          this.connectionStatus$.next(ConnectionStatus.Disconnected)
-          console.log('WebSocket connection closed')
-        },
-      },
-      closingObserver: {
-        next: () => {
-          console.log('Websocket subscription cancelled, websocket closing')
-        },
-      },
     }
     this.websocket$ = webSocket(fullConfig)
     return this.websocket$

client/lib/store.ts

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

client/lib/types.ts

@@ -27,9 +27,3 @@ export interface DBCache<T>{
   sequence: number,
   data: T | { }
 }
-
-export enum ConnectionStatus {
-  Initializing = 'initializing',
-  Connected = 'connected',
-  Disconnected = 'disconnected',
-}