Repository: ecadlabs/taquito (ts)

Issue:
Make PollingSubscribeProvider's polling interval configurable
PollingSubscribeProvider currently uses a hardcoded polling interval(20s) and there's no easy way to change this. This can lead to missing blocks often.
Although it is possible to catch-up missed blocks, this is a pain in the ass to synchronize correctly. So, it would be really helpful to have a property in Config (similar to confirmationPollingIntervalSecond) for this, and I think it would make sense.
"Memory leak" in PollingSubscribeProvider
Because of the way the newBlock$ observable is setup, all blocks are kept in memory because of the publishReplay with an infinite buffer. As each block is roughly 1MB in size, memory consumption can drift quite quickly (standard node.js heap overflows in less than 2 days actually).
NB: memory is actually freed whenever an error occurs calling rpc's getBlock

I'm not too sure about why the publishReplay is there and it seems to me a simple publish() would do the trick. If you really need subscribers to get a value right after subscribing(which could be a value 'in the past' which I don't find relevant in the context), just replacing with publishReplay(1) will also fix the memory issue

PS: i can submit a PR if you want


Pull request:
Polling provider: fixed memory leak replacing publishReplay by publis…
…h and made polling interval configurable
Closes #963 
Closes #943 
Thank you for your contribution to Taquito.

Before submitting this PR, please make sure:

- [ ] Your code builds cleanly without any errors or warnings
- [ ] You have added unit tests
- [ ] You have added integration tests (if relevant/appropriate)
- [ ] All public methods or types have TypeDoc coverage with a complete description, and ideally an @example
- [ ] You have added or updated corresponding documentation
- [ ] If relevant, you have written a first draft summary describing the change for inclusion in Release Notes. 

## Release Note Draft Snippet

__If relevant, please write a summary of your change that will be suitable for
inclusion in the Release Notes for the next Taquito release.__


Patch:
diff --git a/example/example-streamer.ts b/example/example-streamer.ts
index fdd6e6b99..177c3133e 100644
--- a/example/example-streamer.ts
+++ b/example/example-streamer.ts
@@ -3,7 +3,7 @@ import { TezosToolkit } from '@taquito/taquito';
 async function example() {
   const provider = 'https://api.tez.ie/rpc/mainnet';
   const tezos = new TezosToolkit(provider)
-  tezos.setProvider({ config: { shouldObservableSubscriptionRetry: true } });
+  tezos.setProvider({ config: { shouldObservableSubscriptionRetry: true, streamerPollingIntervalMilliseconds: 15000 } });
   try {
 
     const bakerEndorsementFilter = {
diff --git a/packages/taquito/src/context.ts b/packages/taquito/src/context.ts
index d81e381d9..52b1f9487 100644
--- a/packages/taquito/src/context.ts
+++ b/packages/taquito/src/context.ts
@@ -19,28 +19,36 @@ import { Packer } from './packer/interface';
 import { RpcPacker } from './packer/rpc-packer';
 import BigNumber from 'bignumber.js';
 import { retry } from 'rxjs/operators';
-import { OperatorFunction } from 'rxjs';
+import { BehaviorSubject, OperatorFunction } from 'rxjs';
 
 export interface TaquitoProvider<T, K extends Array<any>> {
   new (context: Context, ...rest: K): T;
 }
 
-// The shouldObservableSubscriptionRetry parameter is related to the observable in ObservableSubsription class.
-// When set to true, the observable won't die when getBlock rpc call fails; the error will be reported via the error callback,
-// and it will continue to poll for new blocks.
-export interface Config {
+export interface ConfigConfirmation {
   confirmationPollingIntervalSecond?: number;
   confirmationPollingTimeoutSecond?: number;
   defaultConfirmationCount?: number;
+}
+
+export const defaultConfigConfirmation: Partial<ConfigConfirmation> = {
+  defaultConfirmationCount: 1,
+  confirmationPollingTimeoutSecond: 180
+};
+
+// The shouldObservableSubscriptionRetry parameter is related to the observable in ObservableSubsription class.
+// When set to true, the observable won't die when getBlock rpc call fails; the error will be reported via the error callback,
+// and it will continue to poll for new blocks.
+export interface ConfigStreamer {
+  streamerPollingIntervalMilliseconds?: number;
   shouldObservableSubscriptionRetry?: boolean;
   observableSubscriptionRetryFunction?: OperatorFunction<any, any>;
 }
 
-export const defaultConfig: Partial<Config> = {
-  defaultConfirmationCount: 1,
-  confirmationPollingTimeoutSecond: 180,
+export const defaultConfigStreamer: Required<ConfigStreamer> = {
+  streamerPollingIntervalMilliseconds: 20000,
   shouldObservableSubscriptionRetry: false,
-  observableSubscriptionRetryFunction: retry(),
+  observableSubscriptionRetryFunction: retry()
 };
 
 /**
@@ -66,7 +74,7 @@ export class Context {
     private _rpc: RpcClient | string,
     private _signer: Signer = new NoopSigner(),
     private _proto?: Protocols,
-    private _config?: Partial<Config>,
+    public readonly _config = new BehaviorSubject({...defaultConfigStreamer, ...defaultConfigConfirmation}),
     forger?: Forger,
     injector?: Injector,
     packer?: Packer,
@@ -87,15 +95,16 @@ export class Context {
     this._packer = packer ? packer : new RpcPacker(this);
   }
 
-  get config(): Partial<Config> {
-    return this._config as any;
+  get config(): Partial<ConfigConfirmation> & Required<ConfigStreamer> {
+    return this._config.getValue();
   }
 
-  set config(value: Partial<Config>) {
-    this._config = {
-      ...defaultConfig,
+  set config(value: Partial<ConfigConfirmation> & Required<ConfigStreamer>) {
+    this._config.next({
+      ...defaultConfigConfirmation,
+      ...defaultConfigStreamer,
       ...value,
-    };
+    });
   }
 
   get rpc(): RpcClient {
@@ -224,7 +233,7 @@ export class Context {
       this.rpc,
       this.signer,
       this.proto,
-      this.config,
+      this._config,
       this.forger,
       this._injector,
       this.packer
diff --git a/packages/taquito/src/subscribe/polling-provider.ts b/packages/taquito/src/subscribe/polling-provider.ts
index cd77fb5fd..935d9ca67 100644
--- a/packages/taquito/src/subscribe/polling-provider.ts
+++ b/packages/taquito/src/subscribe/polling-provider.ts
@@ -6,7 +6,7 @@ import {
   first,
   map,
   pluck,
-  publishReplay,
+  publish,
   refCount,
   switchMap,
 } from 'rxjs/operators';
@@ -36,15 +36,20 @@ const applyFilter = (filter: Filter) =>
   });
 
 export class PollingSubscribeProvider implements SubscribeProvider {
-  private newBlock$ = timer(0, this.POLL_INTERVAL).pipe(
+  // Map the changing polling interval to a timer, which will automatically terminate the previous timer when the next one starts.
+  private timer$ = this.context._config.pipe(
+    switchMap((val) => timer(0, val.streamerPollingIntervalMilliseconds)),
+  )
+
+  private newBlock$ = this.timer$.pipe(
     map(() => this.context),
     switchMap(getLastBlock),
     distinctUntilKeyChanged('hash'),
-    publishReplay(),
+    publish(),
     refCount()
   );
 
-  constructor(private context: Context, public readonly POLL_INTERVAL = 20000) {}
+  constructor(private context: Context) { }
 
   subscribe(_filter: 'head'): Subscription<string> {
     return new ObservableSubscription(this.newBlock$.pipe(pluck('hash')), 
diff --git a/packages/taquito/src/taquito.ts b/packages/taquito/src/taquito.ts
index 2a77b67f8..0d21e4320 100644
--- a/packages/taquito/src/taquito.ts
+++ b/packages/taquito/src/taquito.ts
@@ -6,7 +6,7 @@
 import { RpcClient } from '@taquito/rpc';
 import { RPCBatchProvider } from './batch/rpc-batch-provider';
 import { Protocols } from './constants';
-import { Config, Context, TaquitoProvider } from './context';
+import { ConfigConfirmation, ConfigStreamer, Context, TaquitoProvider } from './context';
 import { ContractProvider, EstimationProvider } from './contract/interface';
 import { Extension } from './extension/extension';
 import { Forger } from './forger/interface';
@@ -55,7 +55,7 @@ export interface SetProviderOptions {
   stream?: string | SubscribeProvider;
   signer?: Signer;
   protocol?: Protocols;
-  config?: Config;
+  config?: Partial<ConfigConfirmation> & Partial<ConfigStreamer>;
   packer?: Packer;
 }
 
@@ -124,7 +124,7 @@ export class TezosToolkit {
     this.setPackerProvider(packer);
 
     this._context.proto = protocol;
-    this._context.config = config as Partial<Config>;
+    this._context.config = config as Partial<ConfigConfirmation> & Required<ConfigStreamer>;
   }
 
   /**


Test patch:
diff --git a/packages/taquito/test/operations/batch-operation.spec.ts b/packages/taquito/test/operations/batch-operation.spec.ts
index eae0a18ce..d825f8c72 100644
--- a/packages/taquito/test/operations/batch-operation.spec.ts
+++ b/packages/taquito/test/operations/batch-operation.spec.ts
@@ -1,7 +1,7 @@
 import { ForgedBytes } from '../../src/operations/types';
 import { OperationContentsAndResult } from '@taquito/rpc';
 import { BatchOperation } from '../../src/operations/batch-operation';
-import { defaultConfig } from '../../src/context';
+import { defaultConfigConfirmation, defaultConfigStreamer } from '../../src/context';
 import {
   RevealOperationBuilder,
   TransferOperationBuilder,
@@ -227,7 +227,7 @@ describe('Batch operation', () => {
       rpc: {
         getBlock: jest.fn(),
       },
-      config: { ...defaultConfig },
+      config: { ...defaultConfigConfirmation, ...defaultConfigStreamer },
     };
 
     fakeContext.rpc.getBlock.mockResolvedValue({
diff --git a/packages/taquito/test/operations/delegation-operation.spec.ts b/packages/taquito/test/operations/delegation-operation.spec.ts
index b4faa9d8f..470133dfa 100644
--- a/packages/taquito/test/operations/delegation-operation.spec.ts
+++ b/packages/taquito/test/operations/delegation-operation.spec.ts
@@ -1,9 +1,8 @@
-import { defaultConfig } from '../../src/context';
+import { defaultConfigConfirmation, defaultConfigStreamer } from '../../src/context';
 import { DelegateOperation } from '../../src/operations/delegate-operation';
 import { ForgedBytes } from '../../src/operations/types';
 import {
   RevealOperationBuilder,
-  TransferOperationBuilder,
   DelegationOperationBuilder,
 } from '../helpers';
 
@@ -16,7 +15,7 @@ describe('Delegation operation', () => {
       rpc: {
         getBlock: jest.fn(),
       },
-      config: { ...defaultConfig },
+      config: { ...defaultConfigConfirmation, ...defaultConfigStreamer },
     };
 
     fakeContext.rpc.getBlock.mockResolvedValue({
diff --git a/packages/taquito/test/operations/origination-operation.spec.ts b/packages/taquito/test/operations/origination-operation.spec.ts
index 50e3d0dbd..4590e3efd 100644
--- a/packages/taquito/test/operations/origination-operation.spec.ts
+++ b/packages/taquito/test/operations/origination-operation.spec.ts
@@ -1,4 +1,4 @@
-import { Context, defaultConfig } from '../../src/context';
+import { defaultConfigConfirmation } from '../../src/context';
 import { OriginationOperation } from '../../src/operations/origination-operation';
 import { ForgedBytes } from '../../src/operations/types';
 import { OperationContentsAndResult } from '@taquito/rpc';
@@ -71,7 +71,7 @@ describe('Origination operation', () => {
       rpc: {
         getBlock: jest.fn(),
       },
-      config: { ...defaultConfig },
+      config: { ...defaultConfigConfirmation },
       getConfirmationPollingInterval: jest.fn()
     };
 
diff --git a/packages/taquito/test/operations/transaction-operation.spec.ts b/packages/taquito/test/operations/transaction-operation.spec.ts
index 031425ff2..5eaa877c7 100644
--- a/packages/taquito/test/operations/transaction-operation.spec.ts
+++ b/packages/taquito/test/operations/transaction-operation.spec.ts
@@ -1,7 +1,7 @@
 import { ForgedBytes } from '../../src/operations/types';
 import { OperationContentsAndResult } from '@taquito/rpc';
 import { TransactionOperation } from '../../src/operations/transaction-operation';
-import { defaultConfig } from '../../src/context';
+import { defaultConfigConfirmation, defaultConfigStreamer } from '../../src/context';
 import { TransferOperationBuilder, RevealOperationBuilder } from '../helpers';
 
 describe('Transfer operation', () => {
@@ -85,7 +85,7 @@ describe('Transfer operation', () => {
       rpc: {
         getBlock: jest.fn(),
       },
-      config: { ...defaultConfig },
+      config: { ...defaultConfigConfirmation, ...defaultConfigStreamer },
     };
 
     fakeContext.rpc.getBlock.mockResolvedValue({