Merge branch 'master' into mononaut/scrollable-mempool

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

@@ -1,7 +1,7 @@
 import { Injectable } from '@angular/core';
 import { HttpClient, HttpParams, HttpResponse } from '@angular/common/http';
 import { CpfpInfo, OptimizedMempoolStats, AddressInformation, LiquidPegs, ITranslators,
-  PoolStat, BlockExtended, TransactionStripped, RewardStats, AuditScore, BlockSizesAndWeights, RbfTree } from '../interfaces/node-api.interface';
+  PoolStat, BlockExtended, TransactionStripped, RewardStats, AuditScore, BlockSizesAndWeights, RbfTree, BlockAudit } from '../interfaces/node-api.interface';
 import { Observable } from 'rxjs';
 import { StateService } from './state.service';
 import { WebsocketResponse } from '../interfaces/websocket.interface';
@@ -72,6 +72,10 @@ export class ApiService {
     return this.httpClient.get<OptimizedMempoolStats[]>(this.apiBaseUrl + this.apiBasePath + '/api/v1/statistics/4y');
   }
 
+  listAllTimeStatistics$(): Observable<OptimizedMempoolStats[]> {
+    return this.httpClient.get<OptimizedMempoolStats[]>(this.apiBaseUrl + this.apiBasePath + '/api/v1/statistics/all');
+  }
+
   getTransactionTimes$(txIds: string[]): Observable<number[]> {
     let params = new HttpParams();
     txIds.forEach((txId: string) => {
@@ -245,9 +249,9 @@ export class ApiService {
     );
   }
 
-  getBlockAudit$(hash: string) : Observable<any> {
-    return this.httpClient.get<any>(
-      this.apiBaseUrl + this.apiBasePath + `/api/v1/block/${hash}/audit-summary`, { observe: 'response' }
+  getBlockAudit$(hash: string) : Observable<BlockAudit> {
+    return this.httpClient.get<BlockAudit>(
+      this.apiBaseUrl + this.apiBasePath + `/api/v1/block/${hash}/audit-summary`
     );
   }
 

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

@@ -1,11 +1,11 @@
 import { Inject, Injectable, PLATFORM_ID, LOCALE_ID } from '@angular/core';
-import { ReplaySubject, BehaviorSubject, Subject, fromEvent, Observable } from 'rxjs';
+import { ReplaySubject, BehaviorSubject, Subject, fromEvent, Observable, merge } from 'rxjs';
 import { Transaction } from '../interfaces/electrs.interface';
 import { IBackendInfo, MempoolBlock, MempoolBlockWithTransactions, MempoolBlockDelta, MempoolInfo, Recommendedfees, ReplacedTransaction, TransactionStripped } from '../interfaces/websocket.interface';
 import { BlockExtended, DifficultyAdjustment, MempoolPosition, OptimizedMempoolStats, RbfTree } from '../interfaces/node-api.interface';
 import { Router, NavigationStart } from '@angular/router';
 import { isPlatformBrowser } from '@angular/common';
-import { map, shareReplay } from 'rxjs/operators';
+import { map, scan, shareReplay, tap } from 'rxjs/operators';
 import { StorageService } from './storage.service';
 
 export interface MarkBlockState {
@@ -101,6 +101,7 @@ export class StateService {
   mempoolBlocks$ = new ReplaySubject<MempoolBlock[]>(1);
   mempoolBlockTransactions$ = new Subject<TransactionStripped[]>();
   mempoolBlockDelta$ = new Subject<MempoolBlockDelta>();
+  liveMempoolBlockTransactions$: Observable<{ [txid: string]: TransactionStripped}>;
   txReplaced$ = new Subject<ReplacedTransaction>();
   txRbfInfo$ = new Subject<RbfTree>();
   rbfLatest$ = new Subject<RbfTree[]>();
@@ -167,6 +168,30 @@ export class StateService {
 
     this.blocks$ = new ReplaySubject<[BlockExtended, string]>(this.env.KEEP_BLOCKS_AMOUNT);
 
+    this.liveMempoolBlockTransactions$ = merge(
+      this.mempoolBlockTransactions$.pipe(map(transactions => { return { transactions }; })),
+      this.mempoolBlockDelta$.pipe(map(delta => { return { delta }; })),
+    ).pipe(scan((transactions: { [txid: string]: TransactionStripped }, change: any): { [txid: string]: TransactionStripped } => {
+      if (change.transactions) {
+        const txMap = {}
+        change.transactions.forEach(tx => {
+          txMap[tx.txid] = tx;
+        })
+        return txMap;
+      } else {
+        change.delta.changed.forEach(tx => {
+          transactions[tx.txid].rate = tx.rate;
+        })
+        change.delta.removed.forEach(txid => {
+          delete transactions[txid];
+        });
+        change.delta.added.forEach(tx => {
+          transactions[tx.txid] = tx;
+        });
+        return transactions;
+      }
+    }, {}));
+
     if (this.env.BASE_MODULE === 'bisq') {
       this.network = this.env.BASE_MODULE;
       this.networkChanged$.next(this.env.BASE_MODULE);

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

@@ -235,6 +235,8 @@ export class WebsocketService {
   }
 
   handleResponse(response: WebsocketResponse) {
+    let reinitBlocks = false;
+
     if (response.blocks && response.blocks.length) {
       const blocks = response.blocks;
       let maxHeight = 0;
@@ -256,9 +258,11 @@ export class WebsocketService {
     }
 
     if (response.block) {
-      if (response.block.height > this.stateService.latestBlockHeight) {
+      if (response.block.height === this.stateService.latestBlockHeight + 1) {
         this.stateService.updateChainTip(response.block.height);
         this.stateService.blocks$.next([response.block, response.txConfirmed || '']);
+      } else if (response.block.height > this.stateService.latestBlockHeight + 1) {
+        reinitBlocks = true;
       }
 
       if (response.txConfirmed) {
@@ -369,5 +373,9 @@ export class WebsocketService {
     if (response['git-commit']) {
       this.stateService.backendInfo$.next(response['git-commit']);
     }
+
+    if (reinitBlocks) {
+      this.websocketSubject.next({'refresh-blocks': true});
+    }
   }
 }