add connection monitor to sources

client/lib/patch-db.ts

@@ -1,5 +1,5 @@
 import { merge, Observable, of } from 'rxjs'
-import { concatMap, finalize, tap } from 'rxjs/operators'
+import { concatMap, tap } from 'rxjs/operators'
 import { Source } from './source/source'
 import { Store } from './store'
 import { DBCache, Http } from './types'
@@ -22,4 +22,8 @@ export class PatchDB<T> {
       concatMap(() => of(this.store.cache)),
     )
   }
+
+  connectionMade$ (): Observable<void> {
+    return merge(...this.sources.map(s => s.connectionMade$))
+  }
 }

client/lib/source/poll-source.ts

@@ -1,4 +1,4 @@
-import { BehaviorSubject, concat, from, Observable, of } from 'rxjs'
+import { BehaviorSubject, concat, from, Observable, of, Subject } from 'rxjs'
 import { concatMap, delay, skip, switchMap, take, tap } from 'rxjs/operators'
 import { Store } from '../store'
 import { Http, Update } from '../types'
@@ -9,6 +9,7 @@ export type PollConfig = {
 }
 
 export class PollSource<T> implements Source<T> {
+  connectionMade$ = new Subject<void>()
 
   constructor (
     private readonly pollConfig: PollConfig,
@@ -34,14 +35,15 @@ export class PollSource<T> implements Source<T> {
     const poll$ = concat(updates$, delay$)
 
     return polling$.pipe(
-       switchMap(_ => poll$),
+      switchMap(_ => poll$),
-       concatMap(res => {
+      concatMap(res => {
+        this.connectionMade$.next()
         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>>
         }
-       }),
+      }),
     )
   }
 }

client/lib/source/source.ts

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

client/lib/source/ws-source.ts

@@ -1,10 +1,11 @@
-import { Observable } from 'rxjs'
+import { Observable, Subject } 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<Update<T>> | undefined
+  connectionMade$ = new Subject<void>()
 
   constructor (
     private readonly url: string,
@@ -15,6 +16,7 @@ export class WebsocketSource<T> implements Source<T> {
       url: this.url,
       openObserver: {
         next: () => {
+          this.connectionMade$.next()
           this.websocket$!.next(document.cookie as any)
         },
       },