remove poll altogether

client/index.ts

@@ -1,5 +1,4 @@
 export * from './lib/source/mock-source'
-export * from './lib/source/poll-source'
 export * from './lib/source/ws-source'
 export * from './lib/source/source'
 export * from './lib/json-patch-lib'

client/lib/patch-db.ts

@@ -1,5 +1,5 @@
-import { EMPTY, map, merge, Observable, shareReplay, Subject } from 'rxjs'
+import { map, shareReplay, Subject, timer } from 'rxjs'
-import { catchError, filter, switchMap, tap } from 'rxjs/operators'
+import { catchError, concatMap, tap } from 'rxjs/operators'
 import { Store } from './store'
 import { DBCache, Http } from './types'
 import { Source } from './source/source'
@@ -7,27 +7,21 @@ import { Source } from './source/source'
 export class PatchDB<T> {
   public store: Store<T> = new Store(this.http, this.initialCache)
   public connectionError$ = new Subject<any>()
-  public cache$ = this.sources$.pipe(
+  public cache$ = this.source.watch$(this.store).pipe(
-    switchMap(sources =>
+    catchError((e, watch$) => {
-      merge(...sources.map(s => s.watch$(this.store))).pipe(
-        catchError(e => {
       this.connectionError$.next(e)
-
+      return timer(4000).pipe(concatMap(() => watch$))
-          return EMPTY
     }),
-      ),
+    tap(res => {
-    ),
+      this.connectionError$.next(null)
-    tap(_ => this.connectionError$.next(null)),
-    filter(Boolean),
-    map(res => {
       this.store.update(res)
-      return this.store.cache
     }),
+    map(_ => this.store.cache),
     shareReplay(1),
   )
 
   constructor(
-    private readonly sources$: Observable<Source<T>[]>,
+    private readonly source: Source<T>,
     private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {}

client/lib/source/poll-source.ts

@@ -1,34 +0,0 @@
-import { from, Observable, of, repeat } from 'rxjs'
-import { concatMap, take } from 'rxjs/operators'
-import { Store } from '../store'
-import { Http, Update } from '../types'
-import { Source } from './source'
-
-export type PollConfig = {
-  cooldown: number
-}
-
-export class PollSource<T> implements Source<T> {
-  constructor(
-    private readonly pollConfig: PollConfig,
-    private readonly http: Http<T>,
-  ) {}
-
-  watch$({ sequence$ }: Store<T>): Observable<Update<T> | null> {
-    return sequence$.pipe(
-      concatMap(seq => this.http.getRevisions(seq)),
-      take(1),
-      concatMap(res => {
-        // If Revision[]
-        if (Array.isArray(res)) {
-          // Convert Revision[] it into Observable<Revision> OR return null
-          return res.length ? from(res) : of(null)
-          // If Dump<T>
-        }
-        // Convert Dump<T> into Observable<Dump<T>>
-        return of(res)
-      }),
-      repeat({ delay: this.pollConfig.cooldown }),
-    )
-  }
-}

client/lib/source/source.ts

@@ -3,5 +3,5 @@ import { Store } from '../store'
 import { Update } from '../types'
 
 export interface Source<T> {
-  watch$(store?: Store<T>): Observable<Update<T> | null>
+  watch$(store?: Store<T>): Observable<Update<T>>
 }

client/lib/source/ws-source.ts

@@ -7,6 +7,6 @@ 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: 60000 }))
+    return webSocket<Update<T>>(this.url).pipe(timeout({ first: 21000 }))
   }
 }