Unify grace period config

README.md

@@ -76,8 +76,7 @@ const config = {
     grpcHostnameOverride: "peer0.curator.local"
   },
   stream: {
-    chainInfoPollingIntervalMs: 2000,
-    intervalMs: 1000,
+    gracePeriodMs: 1000,
     batchSize: 10,
     retryOnErrorDelayMs: 5000,
     maxRetryCount: 5

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@gala-chain/stream",
-  "version": "0.0.3",
+  "version": "0.1.0",
   "description": "Streams blocks from GalaChain or Hyperledger Fabric network as RxJS Observables",
   "author": "",
   "private": false,

src/ChainStream.ts

@@ -26,7 +26,7 @@ import {
   tap,
   timer
 } from "rxjs";
-import { bufferCount } from "rxjs/operators";
+import { bufferCount, delay } from "rxjs/operators";
 
 import { CAService, IIdentity } from "./CAService";
 import { ChainService, TransactionFilter } from "./ChainService";
@@ -69,11 +69,11 @@ export class ChainStream {
   }
 
   public startPollingChainHeight(config: {
-    intervalMs: number;
+    gracePeriodMs: number;
     retryOnErrorDelayMs: number;
     maxRetryCount: number;
   }) {
-    return interval(config.intervalMs)
+    return interval(config.gracePeriodMs)
       .pipe(
         switchMap(async () => {
           const info = await this.queryChainInfo();
@@ -102,7 +102,7 @@ export class ChainStream {
 
   public fromBlock(
     startBlock: number,
-    config: { batchSize: number; intervalMs: number; retryOnErrorDelayMs: number; maxRetryCount: number },
+    config: { batchSize: number; retryOnErrorDelayMs: number; maxRetryCount: number; gracePeriodMs: number },
     transactionFilter: TransactionFilter = () => true
   ) {
     let currentBlock = startBlock; // Keep track of the current block we're fetching
@@ -114,8 +114,9 @@ export class ChainStream {
         }
 
         if (currentBlock >= this.chainInfo.value.height) {
-          return timer(config.intervalMs).pipe(
-            tap(() => this.logger.log(`No new blocks, retrying after ${config.intervalMs} ms...`)),
+          return of([]).pipe(
+            tap(() => this.logger.log(`No new blocks, retrying after ${config.gracePeriodMs} ms...`)),
+            delay(config.gracePeriodMs),
             switchMap(() => of([]))
           );
         }
@@ -138,6 +139,8 @@ export class ChainStream {
             return blocks;
           }),
 
+          delay(config.gracePeriodMs),
+
           catchError((err) => {
             const channel = this.chainInfo.value.channelName;
             this.logger.warn(`Error fetching blocks from channel ${channel}: ${err.message}`);

src/ConnectedStream.ts

@@ -21,10 +21,9 @@ import { ConnectedTransactionStream } from "./ConnectedTransactionStream";
 import { Block } from "./types";
 
 export interface StreamConfig {
-  chainInfoPollingIntervalMs: number;
-  intervalMs: number;
-  batchSize: number;
+  gracePeriodMs: number;
   retryOnErrorDelayMs: number;
+  batchSize: number;
   maxRetryCount: number;
 }
 
@@ -36,15 +35,15 @@ export class ConnectedStream {
     private readonly streamConfig: StreamConfig
   ) {
     this.chainHeightSubscription = chainStream.startPollingChainHeight({
-      intervalMs: streamConfig.chainInfoPollingIntervalMs,
-      retryOnErrorDelayMs: streamConfig.retryOnErrorDelayMs,
-      maxRetryCount: streamConfig.maxRetryCount
+      gracePeriodMs: this.streamConfig.gracePeriodMs,
+      retryOnErrorDelayMs: this.streamConfig.retryOnErrorDelayMs,
+      maxRetryCount: this.streamConfig.maxRetryCount
     });
   }
 
   public fromBlock(number: number): Observable<Block> {
     return this.chainStream.fromBlock(number, {
-      intervalMs: this.streamConfig.intervalMs,
+      gracePeriodMs: this.streamConfig.gracePeriodMs,
       batchSize: this.streamConfig.batchSize,
       retryOnErrorDelayMs: this.streamConfig.retryOnErrorDelayMs,
       maxRetryCount: this.streamConfig.maxRetryCount

src/StreamBuilder.ts

@@ -37,8 +37,7 @@ const defaultPeerConfig = {
 };
 
 const defaultStreamConfig = {
-  chainInfoPollingIntervalMs: 2000,
-  intervalMs: 500,
+  gracePeriodMs: 1000,
   batchSize: 10,
   retryOnErrorDelayMs: 5000,
   maxRetryCount: 5
@@ -89,9 +88,7 @@ export class StreamBuilder {
       grpcHostnameOverride: params.peer?.grpcHostnameOverride ?? defaultPeerConfig.grpcHostnameOverride
     };
     this.streamConfig = {
-      chainInfoPollingIntervalMs:
-        params.stream?.chainInfoPollingIntervalMs ?? defaultStreamConfig.chainInfoPollingIntervalMs,
-      intervalMs: params.stream?.intervalMs ?? defaultStreamConfig.intervalMs,
+      gracePeriodMs: params.stream?.gracePeriodMs ?? defaultStreamConfig.gracePeriodMs,
       batchSize: params.stream?.batchSize ?? defaultStreamConfig.batchSize,
       retryOnErrorDelayMs: params.stream?.retryOnErrorDelayMs ?? defaultStreamConfig.retryOnErrorDelayMs,
       maxRetryCount: params.stream?.maxRetryCount ?? defaultStreamConfig.maxRetryCount

src/sample-blocks.ts

@@ -32,8 +32,7 @@ const config = {
     grpcHostnameOverride: "peer0.curator.local"
   },
   stream: {
-    chainInfoPollingIntervalMs: 2000,
-    intervalMs: 1000,
+    gracePeriodMs: 1000,
     batchSize: 10,
     retryOnErrorDelayMs: 5000,
     maxRetryCount: 5

src/sample-transactions.ts

@@ -31,8 +31,7 @@ const config = {
     grpcHostnameOverride: "peer0.curator.local"
   },
   stream: {
-    chainInfoPollingIntervalMs: 2000,
-    intervalMs: 1000,
+    gracePeriodMs: 1000,
     batchSize: 10,
     retryOnErrorDelayMs: 5000,
     maxRetryCount: 5

src/stream.spec.ts

@@ -45,10 +45,9 @@ beforeAll(() => {
     .connect({
       // lower values for faster tests
       stream: {
-        chainInfoPollingIntervalMs: 100,
-        intervalMs: 100,
+        gracePeriodMs: 100,
         batchSize: 3,
-        retryOnErrorDelayMs: 500,
+        retryOnErrorDelayMs: 250,
         maxRetryCount: 10
       },
       logger