chore: fix comments

client/lib/patch-db.ts

@@ -1,4 +1,4 @@
-import { EMPTY, merge, Observable, ReplaySubject, Subject } from 'rxjs'
+import { EMPTY, map, merge, Observable, shareReplay, Subject } from 'rxjs'
 import { catchError, switchMap } from 'rxjs/operators'
 import { Store } from './store'
 import { DBCache, Http } from './types'
@@ -7,30 +7,27 @@ 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$ = new ReplaySubject<DBCache<T>>(1)
+  public cache$ = this.sources$.pipe(
+    switchMap(sources =>
+      merge(...sources.map(s => s.watch$(this.store))).pipe(
+        catchError(e => {
+          this.connectionError$.next(e)
+
+          return EMPTY
+        }),
+      ),
+    ),
+    map(res => {
+      this.store.update(res)
+
+      return this.store.cache
+    }),
+    shareReplay(1),
+  )
 
   constructor(
     private readonly sources$: Observable<Source<T>[]>,
     private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {}
-
-  ngOnInit() {
-    this.sources$
-      .pipe(
-        switchMap(sources =>
-          merge(...sources.map(s => s.watch$(this.store))).pipe(
-            catchError(e => {
-              this.connectionError$.next(e)
-
-              return EMPTY
-            }),
-          ),
-        ),
-      )
-      .subscribe(res => {
-        this.store.update(res)
-        this.cache$.next(this.store.cache)
-      })
-  }
 }

client/lib/source/poll-source.ts

@@ -1,5 +1,5 @@
-import { from, Observable, of, repeatWhen, timer } from 'rxjs'
-import { concatMap, map, take } from 'rxjs/operators'
+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'
@@ -20,7 +20,7 @@ export class PollSource<T> implements Source<T> {
       take(1),
       // convert Revision[] it into Observable<Revision>. Convert Dump<T> into Observable<Dump<T>>
       concatMap(res => (Array.isArray(res) ? from(res) : of(res))),
-      repeatWhen(() => timer(this.pollConfig.cooldown)),
+      repeat({ delay: this.pollConfig.cooldown }),
     )
   }
 }