simplify websocket opening and response types from BE

client/lib/patch-db.ts

@@ -2,43 +2,35 @@ import { EMPTY, merge, Observable, ReplaySubject, Subject } from 'rxjs'
 import { catchError, switchMap } from 'rxjs/operators'
 import { Store } from './store'
 import { DBCache, Http } from './types'
-import { RPCError } from './source/ws-source'
 import { Source } from './source/source'
 
 export class PatchDB<T> {
   public store: Store<T> = new Store(this.http, this.initialCache)
-  public connectionError$ = new Subject<Error>()
+  public connectionError$ = new Subject<any>()
-  public rpcError$ = new Subject<RPCError>()
   public cache$ = new ReplaySubject<DBCache<T>>(1)
 
-  private sub = this.sources$
-    .pipe(
-      switchMap(sources =>
-        merge(...sources.map(s => s.watch$(this.store))).pipe(
-          catchError(e => {
-            this.connectionError$.next(e)
-
-            return EMPTY
-          }),
-        ),
-      ),
-    )
-    .subscribe(res => {
-      if ('result' in res) {
-        this.store.update(res.result)
-        this.cache$.next(this.store.cache)
-      } else {
-        this.rpcError$.next(res)
-      }
-    })
-
   constructor(
     private readonly sources$: Observable<Source<T>[]>,
     private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {}
 
-  clean() {
+  ngOnInit() {
-    this.sub.unsubscribe()
+    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/mock-source.ts

@@ -1,13 +1,11 @@
 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> {
   constructor(private readonly seed: Observable<Update<T>>) {}
 
-  watch$(): Observable<RPCResponse<Update<T>>> {
+  watch$(): Observable<Update<T>> {
-    return this.seed.pipe(map(result => ({ result, jsonrpc: '2.0' })))
+    return this.seed
   }
 }

client/lib/source/poll-source.ts

@@ -3,7 +3,6 @@ import { concatMap, map, take } 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,13 +14,12 @@ export class PollSource<T> implements Source<T> {
     private readonly http: Http<T>,
   ) {}
 
-  watch$({ sequence$ }: Store<T>): Observable<RPCResponse<Update<T>>> {
+  watch$({ sequence$ }: Store<T>): Observable<Update<T>> {
     return sequence$.pipe(
       concatMap(seq => this.http.getRevisions(seq)),
       take(1),
-      // Revision[] converted it into Observable<Revision>, Dump<T> into Observable<Dump<T>>
+      // convert Revision[] it into Observable<Revision>. Convert Dump<T> into Observable<Dump<T>>
       concatMap(res => (Array.isArray(res) ? from(res) : of(res))),
-      map(result => ({ result, jsonrpc: '2.0' as const })),
       repeatWhen(() => timer(this.pollConfig.cooldown)),
     )
   }

client/lib/source/source.ts

@@ -1,8 +1,7 @@
 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<RPCResponse<Update<T>>>
+  watch$(store?: Store<T>): Observable<Update<T>>
 }

client/lib/source/ws-source.ts

@@ -1,54 +1,12 @@
 import { Observable, timeout } from 'rxjs'
-import { webSocket, WebSocketSubject } from 'rxjs/webSocket'
+import { webSocket } from 'rxjs/webSocket'
 import { Update } from '../types'
 import { Source } from './source'
 
 export class WebsocketSource<T> implements Source<T> {
-  constructor(
+  constructor(private readonly url: string) {}
-    private readonly url: string,
-    // TODO: Remove fallback after client app is updated
-    private readonly document: Document = document,
-  ) {}
 
-  watch$(): Observable<RPCResponse<Update<T>>> {
+  watch$(): Observable<Update<T>> {
-    const stream$: WebSocketSubject<RPCResponse<Update<T>>> = webSocket({
+    return webSocket<Update<T>>(this.url).pipe(timeout({ first: 60000 }))
-      url: this.url,
-      openObserver: {
-        next: () => stream$.next(this.document.cookie as any),
-      },
-    })
-
-    return stream$.pipe(timeout({ first: 60000 }))
-  }
-}
-
-interface RPCBase {
-  jsonrpc: '2.0'
-}
-
-export interface RPCSuccess<T> extends RPCBase {
-  result: T
-}
-
-export interface RPCError extends RPCBase {
-  error: {
-    code: number // 34 means unauthenticated
-    message: string
-    data: {
-      details: string
-    }
-  }
-}
-export type RPCResponse<T> = RPCSuccess<T> | RPCError
-
-class RpcError {
-  code: number
-  message: string
-  details: string
-
-  constructor(e: RPCError['error']) {
-    this.code = e.code
-    this.message = e.message
-    this.details = e.data.details
   }
 }