polling working without refresh

client/lib/patch-db.ts

@@ -1,26 +1,46 @@
-import { merge, Observable, of } from 'rxjs'
+import { merge, Observable, Subject, Subscription } from 'rxjs'
-import { concatMap, finalize, tap } from 'rxjs/operators'
-import { Source } from './source/source'
 import { Store } from './store'
 import { DBCache, Http } from './types'
+import { RPCError } from './source/ws-source'
+import { Source } from './source/source'
 
 export class PatchDB<T> {
-  store: Store<T>
+  public store: Store<T> = new Store(this.http, this.initialCache)
+  public connectionError$ = new Subject<Error>()
+  public rpcError$ = new Subject<RPCError>()
+  public cache$ = new Subject<DBCache<T>>()
+
+  private updatesSub?: Subscription
+  private sourcesSub = this.sources$.subscribe(sources => {
+    this.updatesSub = merge(...sources.map(s => s.watch$(this.store))).subscribe({
+      next: (res) => {
+        console.log('PatchDB -> subscribedChanges -> JCWM:', { res })
+        if ('result' in res) {
+          this.store.update(res.result)
+          this.cache$.next(this.store.cache)
+        }
+        else {
+          this.rpcError$.next(res)
+        }
+      },
+      error: (e) => {
+        this.connectionError$.next(e)
+      },
+    })
+  })
 
   constructor (
-    private readonly sources: Source<T>[],
+    private readonly sources$: Observable<Source<T>[]>,
     private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {
-    this.store = new Store(this.http, this.initialCache)
+    console.log('STARTING PATCH 1')
   }
 
-  sync$ (): Observable<DBCache<T>> {
+  clean () {
-    return merge(...this.sources.map(s => s.watch$(this.store)))
+    this.sourcesSub.unsubscribe()
-    .pipe(
+    if (this.updatesSub) {
-      tap(update => this.store.update(update)),
+      this.updatesSub.unsubscribe()
-      concatMap(() => of(this.store.cache)),
+    }
-      finalize(() => this.store.reset()),
-    )
   }
 }

client/lib/source/mock-source.ts

@@ -1,6 +1,8 @@
 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> {
 
@@ -8,7 +10,8 @@ export class MockSource<T> implements Source<T> {
     private readonly seed: Observable<Update<T>>,
   ) { }
 
-  watch$ (): Observable<Update<T>> {
+  watch$ (): Observable<RPCResponse<Update<T>>> {
-    return this.seed
+    return this.seed.pipe(map(result => ({ result,
-  }
+      jsonrpc: '2.0' })))
+    }
 }

client/lib/source/poll-source.ts

@@ -1,8 +1,9 @@
 import { BehaviorSubject, concat, from, Observable, of, Subject } from 'rxjs'
-import { concatMap, delay, skip, switchMap, take, tap } from 'rxjs/operators'
+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
@@ -15,7 +16,7 @@ export class PollSource<T> implements Source<T> {
     private readonly http: Http<T>,
   ) { }
 
-  watch$ (store: Store<T>): Observable<Update<T>> {
+  watch$ (store: Store<T>): Observable<RPCResponse<Update<T>>> {
     const polling$ = new BehaviorSubject('')
 
     const updates$ = of({ })
@@ -42,6 +43,8 @@ export class PollSource<T> implements Source<T> {
           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,7 +1,8 @@
 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<Update<T>>
+  watch$ (store?: Store<T>): Observable<RPCResponse<Update<T>>>
 }

client/lib/source/ws-source.ts

@@ -1,5 +1,4 @@
 import { Observable } from 'rxjs'
-import { map } from 'rxjs/operators'
 import { webSocket, WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket'
 import { Update } from '../types'
 import { Source } from './source'
@@ -11,7 +10,7 @@ export class WebsocketSource<T> implements Source<T> {
     private readonly url: string,
   ) { }
 
-  watch$ (): Observable<Update<T>> {
+  watch$ (): Observable<RPCResponse<Update<T>>> {
     const fullConfig: WebSocketSubjectConfig<RPCResponse<Update<T>>> = {
       url: this.url,
       openObserver: {
@@ -21,12 +20,7 @@ export class WebsocketSource<T> implements Source<T> {
       },
     }
     this.websocket$ = webSocket(fullConfig)
-    return this.websocket$.pipe(
+    return this.websocket$
-      map(res => {
-        if (isRpcSuccess(res)) return res.result
-        if (isRpcError(res)) throw new RpcError(res.error)
-      }),
-    ) as Observable<Update<T>>
   }
 }
 
@@ -47,7 +41,6 @@ export interface RPCError extends RPCBase {
     }
   }
 }
-
 export type RPCResponse<T> = RPCSuccess<T> | RPCError
 
 function isRpcError<Error, Result> (arg: { error: Error } | { result: Result}): arg is { error: Error } {