add observConnected to RxrpcInvoker and RxrpcClient (#7)

Signed-off-by: koazu1 <[email protected]>

Co-authored-by: koazu1 <[email protected]>

Author: urikoa

src/lib/rxrpc-client.ts

@@ -1,5 +1,5 @@
 import {defer, interval, Observable, of, OperatorFunction, Subject, throwError} from 'rxjs';
-import {finalize, flatMap, shareReplay, takeUntil, takeWhile} from 'rxjs/operators'
+import {distinctUntilChanged, finalize, flatMap, shareReplay, takeUntil, takeWhile} from 'rxjs/operators'
 import {Response} from './data/response';
 import {Result} from './data/result';
 import {Invocation, Invocations} from './data/invocation';
@@ -30,6 +30,7 @@ export class RxRpcClient extends RxRpcInvoker {
     private listeners: RxRpcInvocationListener[] = [];
     private currentConnection: RxRpcConnection;
     private readonly sharedInvocations = new Map<string, Observable<Result>>();
+    private readonly connectedSubject = new Subject<boolean>();
 
     constructor(private readonly transport: RxRpcTransport, options?: RxRpcClientOptions) {
         super();
@@ -54,6 +55,10 @@ export class RxRpcClient extends RxRpcInvoker {
             });
     }
 
+    public observeConnected(): Observable<boolean> {
+        return this.connectedSubject.pipe(distinctUntilChanged());
+    }
+
     public addListener(listener: RxRpcInvocationListener): RxRpcInvocationListenerSubscription {
         this.listeners.push(listener);
         return { unsubscribe: () => this.listeners = this.listeners.filter(l => l != listener) };
@@ -138,6 +143,7 @@ export class RxRpcClient extends RxRpcInvoker {
 
     private onConnected(connection: RxRpcConnection) {
         this.currentConnection = connection;
+        this.connectedSubject.next(true);
         this.currentConnection.messages
             .pipe(takeUntil(this.cancelledSubject))
             .subscribe(
@@ -156,12 +162,15 @@ export class RxRpcClient extends RxRpcInvoker {
         this.invocations.clear();
         this.cancelledSubject.next();
         this.currentConnection = null;
+        this.connectedSubject.next(false);
     }
 
     private dispatchResponse(response: Response) {
         this.listeners.forEach(l => l.onResponse(response));
-        this.invocations
-            .get(response.invocationId)
-            .next(response.result);
+        if (this.invocations && this.invocations.has(response.invocationId)) {
+            this.invocations
+                .get(response.invocationId)
+                .next(response.result);
+        }
     }
 }

src/lib/rxrpc-invoker.ts

@@ -3,4 +3,5 @@ import {Observable} from 'rxjs';
 export abstract class RxRpcInvoker {
     public abstract invoke<T>(method: string, args: any): Observable<T>;
     public abstract invokeShared<T>(method: string, replayCount: number, args: any): Observable<T>;
+    public abstract observeConnected(): Observable<boolean>;
 }