mirror of
https://github.com/Start9Labs/patch-db.git
synced 2026-03-26 02:11:54 +00:00
update WS for unauthorized response
This commit is contained in:
committed by
Aiden McClelland
parent
afbdab4a45
commit
7a81c7da57
@@ -1,17 +1,18 @@
|
|||||||
import { Observable, Subject } from 'rxjs'
|
import { Observable } from 'rxjs'
|
||||||
|
import { map } from 'rxjs/operators'
|
||||||
import { webSocket, WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket'
|
import { webSocket, WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket'
|
||||||
import { Update } from '../types'
|
import { Update } from '../types'
|
||||||
import { Source } from './source'
|
import { Source } from './source'
|
||||||
|
|
||||||
export class WebsocketSource<T> implements Source<T> {
|
export class WebsocketSource<T> implements Source<T> {
|
||||||
private websocket$: WebSocketSubject<Update<T>> | undefined
|
private websocket$: WebSocketSubject<RPCResponse<Update<T>>> | undefined
|
||||||
|
|
||||||
constructor (
|
constructor (
|
||||||
private readonly url: string,
|
private readonly url: string,
|
||||||
) { }
|
) { }
|
||||||
|
|
||||||
watch$ (): Observable<Update<T>> {
|
watch$ (): Observable<Update<T>> {
|
||||||
const fullConfig: WebSocketSubjectConfig<Update<T>> = {
|
const fullConfig: WebSocketSubjectConfig<RPCResponse<Update<T>>> = {
|
||||||
url: this.url,
|
url: this.url,
|
||||||
openObserver: {
|
openObserver: {
|
||||||
next: () => {
|
next: () => {
|
||||||
@@ -20,6 +21,51 @@ export class WebsocketSource<T> implements Source<T> {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
this.websocket$ = webSocket(fullConfig)
|
this.websocket$ = webSocket(fullConfig)
|
||||||
return this.websocket$
|
return this.websocket$.pipe(
|
||||||
|
map(res => {
|
||||||
|
if (isRpcSuccess(res)) return res.result
|
||||||
|
if (isRpcError(res)) throw new RpcError(res.error)
|
||||||
|
}),
|
||||||
|
) as Observable<Update<T>>
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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
|
||||||
|
|
||||||
|
function isRpcError<Error, Result> (arg: { error: Error } | { result: Result}): arg is { error: Error } {
|
||||||
|
return !!(arg as any).error
|
||||||
|
}
|
||||||
|
|
||||||
|
function isRpcSuccess<Error, Result> (arg: { error: Error } | { result: Result}): arg is { result: Result } {
|
||||||
|
return !!(arg as any).result
|
||||||
|
}
|
||||||
|
|
||||||
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user