Refactored frontend data handling.

frontend/src/app/services/api.service.ts

@@ -1,9 +1,10 @@
 import { Injectable } from '@angular/core';
-import { environment } from '../../environments/environment';
 import { webSocket } from 'rxjs/webSocket';
 import { HttpClient, HttpParams } from '@angular/common/http';
-import { IMempoolDefaultResponse, IMempoolStats, IBlockTransaction } from '../blockchain/interfaces';
+import { IMempoolDefaultResponse, IMempoolStats, IBlockTransaction, IBlock } from '../blockchain/interfaces';
 import { Observable } from 'rxjs';
+import { MemPoolService } from './mem-pool.service';
+import { tap, retryWhen } from 'rxjs/operators';
 
 const WEB_SOCKET_URL = 'ws://' + document.location.hostname + ':8999';
 const API_BASE_URL = '/api/v1';
@@ -12,11 +13,95 @@ const API_BASE_URL = '/api/v1';
   providedIn: 'root'
 })
 export class ApiService {
+  private websocketSubject: Observable<IMempoolDefaultResponse> = webSocket<IMempoolDefaultResponse | any>(WEB_SOCKET_URL)
+
   constructor(
     private httpClient: HttpClient,
-  ) { }
+    private memPoolService: MemPoolService,
+  ) {
+    this.startSubscription();
+  }
 
-  websocketSubject = webSocket<IMempoolDefaultResponse>(WEB_SOCKET_URL);
+  startSubscription() {
+    this.websocketSubject
+      .pipe(
+        retryWhen((errors: any) => errors
+          .pipe(
+            tap(() => this.memPoolService.isOffline$.next(true))
+          )
+        ),
+      )
+      .subscribe((response: IMempoolDefaultResponse) => {
+          this.memPoolService.isOffline$.next(false);
+
+          if (response.blocks && response.blocks.length) {
+            const blocks = response.blocks;
+            // blocks.reverse();
+            blocks.forEach((block: IBlock) => this.memPoolService.blocks$.next(block));
+          }
+          if (response.block) {
+            this.memPoolService.blocks$.next(response.block);
+          }
+
+          if (response.projectedBlocks) {
+            this.memPoolService.projectedBlocks$.next(response.projectedBlocks);
+          }
+
+          if (response.mempoolInfo && response.txPerSecond !== undefined) {
+            this.memPoolService.mempoolStats$.next({
+              memPoolInfo: response.mempoolInfo,
+              txPerSecond: response.txPerSecond,
+              vBytesPerSecond: response.vBytesPerSecond,
+            });
+          }
+
+          if (response.conversions) {
+            this.memPoolService.conversions$.next(response.conversions);
+          }
+
+          if (response.projectedBlocks) {
+            const mempoolWeight = response.projectedBlocks.map((block: any) => block.blockWeight).reduce((a: any, b: any) => a + b);
+            this.memPoolService.mempoolWeight$.next(mempoolWeight);
+          }
+
+          if (response['track-tx']) {
+            let txTrackingEnabled;
+            let txTrackingBlockHeight;
+            let txTrackingTx = null;
+            let txShowTxNotFound = false;
+            if (response['track-tx'].tracking) {
+              txTrackingEnabled = true;
+              txTrackingBlockHeight = response['track-tx'].blockHeight;
+              if (response['track-tx'].tx) {
+                txTrackingTx = response['track-tx'].tx;
+              }
+            } else {
+              txTrackingEnabled = false;
+              txTrackingTx = null;
+              txTrackingBlockHeight = 0;
+            }
+            if (response['track-tx'].message && response['track-tx'].message === 'not-found') {
+              txShowTxNotFound = true;
+            }
+            this.memPoolService.txTracking$.next({
+              enabled: txTrackingEnabled,
+              tx: txTrackingTx,
+              blockHeight: txTrackingBlockHeight,
+              notFound: txShowTxNotFound,
+            });
+          }
+        }),
+        (err: Error) => {
+          console.log(err);
+          console.log('Error, retrying in 10 sec');
+          setTimeout(() => this.startSubscription(), 10000);
+        };
+  }
+
+  sendWebSocket(data: any) {
+    // @ts-ignore
+    this.websocketSubject.next(data);
+  }
 
   listTransactionsForBlock$(height: number): Observable<IBlockTransaction[]> {
     return this.httpClient.get<IBlockTransaction[]>(API_BASE_URL + '/transactions/height/' + height);

frontend/src/app/services/mem-pool.service.ts

@@ -1,20 +1,35 @@
 import { Injectable } from '@angular/core';
-import { Subject, ReplaySubject } from 'rxjs';
-import { IMempoolInfo } from '../blockchain/interfaces';
+import { ReplaySubject, BehaviorSubject } from 'rxjs';
+import { IMempoolInfo, IBlock, IProjectedBlock, ITransaction } from '../blockchain/interfaces';
 
-export interface MemPoolState {
+export interface IMemPoolState {
   memPoolInfo: IMempoolInfo;
   txPerSecond: number;
   vBytesPerSecond: number;
 }
 
+export interface ITxTracking {
+  enabled: boolean;
+  tx: ITransaction | null;
+  blockHeight: number;
+  notFound: boolean;
+}
+
 @Injectable({
   providedIn: 'root'
 })
 export class MemPoolService {
-  loaderSubject = new Subject<MemPoolState>();
-  isOffline = new Subject<boolean>();
-  txIdSearch = new Subject<string>();
-  conversions = new ReplaySubject<any>();
-  mempoolWeight = new Subject<number>();
+  mempoolStats$ = new ReplaySubject<IMemPoolState>();
+  isOffline$ = new ReplaySubject<boolean>();
+  txIdSearch$ = new ReplaySubject<string>();
+  conversions$ = new ReplaySubject<any>();
+  mempoolWeight$ = new ReplaySubject<number>();
+  txTracking$ = new BehaviorSubject<ITxTracking>({
+    enabled: false,
+    tx: null,
+    blockHeight: 0,
+    notFound: false,
+  });
+  blocks$ = new ReplaySubject<IBlock>(8);
+  projectedBlocks$ = new BehaviorSubject<IProjectedBlock[]>([]);
 }