refactor!: rename 'coldObservableProvider' util to 'poll'

this util is not Provider-aware
it resides in 'util-rxjs' package, which does not depend on 'core'

follow-up commit will add a new Provider-aware util that
knows about retryable ProviderErrors

BREAKING CHANGE: rename poll props 'provider' to 'sample'

Author: mkazlauskas

packages/tx-construction/src/tx-builder/TxBuilder.ts

@@ -39,11 +39,11 @@ import {
   validateValidityInterval
 } from './utils';
 import { SelectionSkeleton } from '@cardano-sdk/input-selection';
-import { coldObservableProvider } from '@cardano-sdk/util-rxjs';
 import { contextLogger, deepEquals } from '@cardano-sdk/util';
 import { createOutputValidator } from '../output-validation';
 import { initializeTx } from './initializeTx';
 import { lastValueFrom } from 'rxjs';
+import { poll } from '@cardano-sdk/util-rxjs';
 import omit from 'lodash/omit.js';
 import uniq from 'lodash/uniq.js';
 
@@ -499,12 +499,12 @@ export class GenericTxBuilder implements TxBuilder {
       allRewardAccounts = uniq([...knownAddresses, ...newAddresses]).map(({ rewardAccount }) => rewardAccount);
     }
 
-    const rewardAccounts$ = coldObservableProvider({
+    const rewardAccounts$ = poll({
       logger: contextLogger(this.#logger, 'getOrCreateRewardAccounts'),
       pollUntil: (rewardAccounts) =>
         allRewardAccounts.every((newAccount) => rewardAccounts.some((acct) => acct.address === newAccount)),
-      provider: this.#dependencies.txBuilderProviders.rewardAccounts,
-      retryBackoffConfig: { initialInterval: 10, maxInterval: 100, maxRetries: 10 }
+      retryBackoffConfig: { initialInterval: 10, maxInterval: 100, maxRetries: 10 },
+      sample: this.#dependencies.txBuilderProviders.rewardAccounts
     });
 
     try {

packages/util-rxjs/src/index.ts

@@ -5,5 +5,5 @@ export * from './passthrough';
 export * from './finalizeWithLatest';
 export * from './concatAndCombineLatest';
 export * from './shareRetryBackoff';
-export * from './coldObservableProvider';
+export * from './poll';
 export * from './types';

packages/util-rxjs/src/coldObservableProvider.ts renamed to packages/util-rxjs/src/poll.ts

@@ -18,8 +18,8 @@ import {
 } from 'rxjs';
 import { RetryBackoffConfig, retryBackoff } from 'backoff-rxjs';
 
-export interface ColdObservableProviderProps<T> {
-  provider: () => Promise<T>;
+export interface PollProps<T> {
+  sample: () => Promise<T>;
   retryBackoffConfig: RetryBackoffConfig;
   onFatalError?: (value: unknown) => void;
   trigger$?: Observable<unknown>;
@@ -30,8 +30,8 @@ export interface ColdObservableProviderProps<T> {
   logger: Logger;
 }
 
-export const coldObservableProvider = <T>({
-  provider,
+export const poll = <T>({
+  sample,
   retryBackoffConfig,
   onFatalError,
   trigger$ = of(true),
@@ -40,15 +40,15 @@ export const coldObservableProvider = <T>({
   cancel$ = NEVER,
   pollUntil = () => true,
   logger
-}: ColdObservableProviderProps<T>) =>
+}: PollProps<T>) =>
   new Observable<T>((subscriber) => {
     const cancelOnFatalError$ = new Subject<boolean>();
     const internalCancel$ = merge(cancel$, cancelOnFatalError$);
     const sub = trigger$
       .pipe(
         combinator(() =>
           defer(() =>
-            from(provider()).pipe(
+            from(sample()).pipe(
               mergeMap((v) =>
                 pollUntil(v)
                   ? of(v)

packages/util-rxjs/test/coldObservableProvider.test.ts renamed to packages/util-rxjs/test/poll.test.ts

@@ -2,84 +2,84 @@ import { BehaviorSubject, EmptyError, Subject, firstValueFrom, lastValueFrom, ta
 import { InvalidStringError } from '@cardano-sdk/util';
 import { RetryBackoffConfig, retryBackoff } from 'backoff-rxjs';
 import { TestLogger, createLogger } from '@cardano-sdk/util-dev';
-import { coldObservableProvider } from '../src';
+import { poll } from '../src';
 
 // There might be a more elegant way to mock with original implementation (spy)
 jest.mock('backoff-rxjs', () => ({
   retryBackoff: jest.fn().mockImplementation((...args) => jest.requireActual('backoff-rxjs').retryBackoff(...args))
 }));
 
-describe('coldObservableProvider', () => {
+describe('poll', () => {
   let logger: TestLogger;
   const testErrorStr = 'Test error';
 
   beforeEach(() => {
     logger = createLogger({ record: true });
   });
 
-  it('returns an observable that calls underlying provider on each subscription and uses retryBackoff', async () => {
-    const underlyingProvider = jest.fn().mockResolvedValue(true);
+  it('returns an observable that calls sample on each subscription and uses retryBackoff', async () => {
+    const sample = jest.fn().mockResolvedValue(true);
     const backoffConfig: RetryBackoffConfig = { initialInterval: 1 };
-    const provider$ = coldObservableProvider({
+    const values$ = poll({
       logger,
-      provider: underlyingProvider,
-      retryBackoffConfig: backoffConfig
+      retryBackoffConfig: backoffConfig,
+      sample
     });
-    expect(await firstValueFrom(provider$)).toBe(true);
-    expect(await firstValueFrom(provider$)).toBe(true);
-    expect(underlyingProvider).toBeCalledTimes(2);
+    expect(await firstValueFrom(values$)).toBe(true);
+    expect(await firstValueFrom(values$)).toBe(true);
+    expect(sample).toBeCalledTimes(2);
     expect(retryBackoff).toBeCalledTimes(2);
   });
 
-  it('provider is unsubscribed on cancel emit', async () => {
-    const fakeProviderSubject = new Subject();
-    const underlyingProvider = () => firstValueFrom(fakeProviderSubject);
+  it('completes on cancel emit', async () => {
+    const fakeSampleSubject = new Subject();
+    const sample = () => firstValueFrom(fakeSampleSubject);
     const backoffConfig: RetryBackoffConfig = { initialInterval: 1 };
     const cancel$ = new BehaviorSubject<boolean>(true);
-    const provider$ = coldObservableProvider({
+    const values$ = poll({
       cancel$,
       logger,
-      provider: underlyingProvider,
-      retryBackoffConfig: backoffConfig
+      retryBackoffConfig: backoffConfig,
+      sample
     });
 
     try {
-      await firstValueFrom(provider$);
+      await firstValueFrom(values$);
     } catch (error) {
       expect(error).toBeInstanceOf(EmptyError);
     }
     expect.assertions(1);
   });
 
-  it('retries using retryBackoff, when underlying provider rejects', async () => {
-    const underlyingProvider = jest.fn().mockRejectedValueOnce(false).mockResolvedValue(true);
+  it('retries using retryBackoff, when sample rejects', async () => {
+    const sample = jest.fn().mockRejectedValueOnce(false).mockResolvedValue(true);
     const retryBackoffConfig: RetryBackoffConfig = { initialInterval: 1 };
-    const provider$ = coldObservableProvider({ logger, provider: underlyingProvider, retryBackoffConfig });
-    const resolvedValue = await firstValueFrom(provider$);
-    expect(underlyingProvider).toBeCalledTimes(2);
+    const values$ = poll({ logger, retryBackoffConfig, sample });
+    const resolvedValue = await firstValueFrom(values$);
+    expect(sample).toBeCalledTimes(2);
     expect(resolvedValue).toBeTruthy();
   });
 
-  it('does not retry, when underlying provider rejects with InvalidStringError', async () => {
+  it('does not retry, when sample rejects with InvalidStringError', async () => {
     const testValue = { test: 'value' };
     const testError = new InvalidStringError('Test invalid string error');
-    const underlyingProvider = jest
+    const sample = jest
       .fn()
       .mockRejectedValueOnce(new Error(testErrorStr))
       .mockResolvedValueOnce(testValue)
       .mockRejectedValueOnce(testError)
       .mockResolvedValueOnce(testValue);
     const onFatalError = jest.fn();
     const retryBackoffConfig: RetryBackoffConfig = { initialInterval: 1, shouldRetry: () => true };
-    const provider$ = coldObservableProvider({
+    const values$ = poll({
       logger,
       onFatalError,
-      provider: underlyingProvider,
-      retryBackoffConfig
+      retryBackoffConfig,
+      sample
     });
-    await expect(firstValueFrom(provider$)).resolves.toBe(testValue);
-    await expect(firstValueFrom(provider$)).rejects.toThrow(EmptyError);
-    expect(underlyingProvider).toBeCalledTimes(3);
+    await expect(firstValueFrom(values$)).resolves.toBe(testValue);
+    await expect(firstValueFrom(values$)).rejects.toThrow(EmptyError);
+    expect(sample).toBeCalledTimes(3);
     expect(onFatalError).toBeCalledWith(testError);
     expect(logger.messages).toStrictEqual([
       { level: 'error', message: [new Error(testErrorStr)] },
@@ -89,45 +89,45 @@ describe('coldObservableProvider', () => {
     ]);
   });
 
-  it('polls the provider until the pollUntil condition is satisfied', async () => {
-    const underlyingProvider = jest
+  it('polls sample until the pollUntil condition is satisfied', async () => {
+    const sample = jest
       .fn()
       .mockResolvedValueOnce('a')
       .mockResolvedValueOnce('b')
       .mockResolvedValueOnce('c')
       .mockResolvedValue('Never reached');
     const backoffConfig: RetryBackoffConfig = { initialInterval: 1 };
 
-    const provider$ = coldObservableProvider({
+    const values$ = poll({
       logger,
       pollUntil: (v) => v === 'c',
-      provider: underlyingProvider,
-      retryBackoffConfig: backoffConfig
+      retryBackoffConfig: backoffConfig,
+      sample
     });
 
-    const providerValues: unknown[] = [];
-    await lastValueFrom(provider$.pipe(tap((v) => providerValues.push(v))));
-    expect(providerValues).toEqual(['a', 'b', 'c']);
-    expect(underlyingProvider).toBeCalledTimes(3);
+    const sampleValues: unknown[] = [];
+    await lastValueFrom(values$.pipe(tap((v) => sampleValues.push(v))));
+    expect(sampleValues).toEqual(['a', 'b', 'c']);
+    expect(sample).toBeCalledTimes(3);
   });
 
   it('stops retrying after maxRetries attempts and handles the error in catchError', async () => {
     const testError = new Error(testErrorStr);
-    const underlyingProvider = jest.fn().mockRejectedValue(testError);
+    const sample = jest.fn().mockRejectedValue(testError);
     const maxRetries = 3;
     const retryBackoffConfig: RetryBackoffConfig = { initialInterval: 1, maxRetries };
     const onFatalError = jest.fn();
 
-    const provider$ = coldObservableProvider({
+    const values$ = poll({
       logger,
       onFatalError,
-      provider: underlyingProvider,
-      retryBackoffConfig
+      retryBackoffConfig,
+      sample
     });
 
-    await expect(firstValueFrom(provider$)).rejects.toThrow(testError);
+    await expect(firstValueFrom(values$)).rejects.toThrow(testError);
 
-    expect(underlyingProvider).toBeCalledTimes(maxRetries + 1);
+    expect(sample).toBeCalledTimes(maxRetries + 1);
     expect(onFatalError).toBeCalledWith(expect.any(Error));
     expect(logger.messages).toStrictEqual([
       { level: 'error', message: [testError] },

packages/wallet/src/Wallets/BaseWallet.ts

@@ -71,7 +71,7 @@ import {
   TxSubmitProvider,
   UtxoProvider
 } from '@cardano-sdk/core';
-import { BehaviorObservable, TrackerSubject, coldObservableProvider } from '@cardano-sdk/util-rxjs';
+import { BehaviorObservable, TrackerSubject, poll } from '@cardano-sdk/util-rxjs';
 import {
   BehaviorSubject,
   EMPTY,
@@ -369,15 +369,15 @@ export class BaseWallet implements ObservableWallet {
 
     if (isBip32PublicCredentialsManager(this.#publicCredentialsManager)) {
       this.#addressTracker = createAddressTracker({
-        addressDiscovery$: coldObservableProvider({
+        addressDiscovery$: poll({
           cancel$,
           logger: contextLogger(this.#logger, 'addressDiscovery$'),
           onFatalError,
-          provider: () => {
+          retryBackoffConfig,
+          sample: () => {
             const credManager = this.#publicCredentialsManager as Bip32PublicCredentialsManager;
             return credManager.addressDiscovery.discover(credManager.bip32Account);
-          },
-          retryBackoffConfig
+          }
         }).pipe(
           take(1),
           catchError((error) => {
@@ -403,12 +403,12 @@ export class BaseWallet implements ObservableWallet {
       logger: contextLogger(this.#logger, 'tip$'),
       maxPollInterval: maxInterval,
       minPollInterval: pollInterval,
-      provider$: coldObservableProvider({
+      provider$: poll({
         cancel$,
         logger: contextLogger(this.#logger, 'tip$'),
         onFatalError,
-        provider: this.networkInfoProvider.ledgerTip,
-        retryBackoffConfig
+        retryBackoffConfig,
+        sample: this.networkInfoProvider.ledgerTip
       }),
       store: stores.tip,
       syncStatus: this.syncStatus
@@ -426,13 +426,13 @@ export class BaseWallet implements ObservableWallet {
     // Era summaries
     const eraSummariesTrigger = new BehaviorSubject<void>(void 0);
     this.eraSummaries$ = new PersistentDocumentTrackerSubject(
-      coldObservableProvider({
+      poll({
         cancel$,
         equals: deepEquals,
         logger: contextLogger(this.#logger, 'eraSummaries$'),
         onFatalError,
-        provider: this.networkInfoProvider.eraSummaries,
         retryBackoffConfig,
+        sample: this.networkInfoProvider.eraSummaries,
         trigger$: eraSummariesTrigger.pipe(tap(() => 'Trigger request era summaries'))
       }),
       stores.eraSummaries
@@ -450,25 +450,25 @@ export class BaseWallet implements ObservableWallet {
       tap((epoch) => this.#logger.debug(`Current epoch is ${epoch}`))
     );
     this.protocolParameters$ = new PersistentDocumentTrackerSubject(
-      coldObservableProvider({
+      poll({
         cancel$,
         equals: isEqual,
         logger: contextLogger(this.#logger, 'protocolParameters$'),
         onFatalError,
-        provider: this.networkInfoProvider.protocolParameters,
         retryBackoffConfig,
+        sample: this.networkInfoProvider.protocolParameters,
         trigger$: epoch$
       }),
       stores.protocolParameters
     );
     this.genesisParameters$ = new PersistentDocumentTrackerSubject(
-      coldObservableProvider({
+      poll({
         cancel$,
         equals: isEqual,
         logger: contextLogger(this.#logger, 'genesisParameters$'),
         onFatalError,
-        provider: this.networkInfoProvider.genesisParameters,
         retryBackoffConfig,
+        sample: this.networkInfoProvider.genesisParameters,
         trigger$: epoch$
       }),
       stores.genesisParameters
@@ -602,13 +602,13 @@ export class BaseWallet implements ObservableWallet {
     this.handles$ = this.handleProvider
       ? this.initializeHandles(
           new PersistentDocumentTrackerSubject(
-            coldObservableProvider({
+            poll({
               cancel$,
               equals: isEqual,
               logger: contextLogger(this.#logger, 'handles$'),
               onFatalError,
-              provider: () => this.handleProvider.getPolicyIds(),
-              retryBackoffConfig
+              retryBackoffConfig,
+              sample: () => this.handleProvider.getPolicyIds()
             }),
             stores.policyIds
           )

packages/wallet/src/services/AssetsTracker.ts

@@ -20,7 +20,7 @@ import {
 } from 'rxjs';
 import { RetryBackoffConfig } from 'backoff-rxjs';
 import { TrackedAssetProvider } from './ProviderTracker';
-import { coldObservableProvider, concatAndCombineLatest } from '@cardano-sdk/util-rxjs';
+import { concatAndCombineLatest, poll } from '@cardano-sdk/util-rxjs';
 import { deepEquals, isNotNil } from '@cardano-sdk/util';
 import { newTransactions$ } from './TransactionsTracker';
 import chunk from 'lodash/chunk.js';
@@ -139,13 +139,13 @@ export const createAssetService =
   (assetIds: Cardano.AssetId[]) =>
     concatAndCombineLatest(
       chunk(assetIds, ASSET_INFO_FETCH_CHUNK_SIZE).map((assetIdsChunk) =>
-        coldObservableProvider({
+        poll({
           logger,
           onFatalError,
           pollUntil: isEveryAssetInfoComplete,
-          provider: () =>
-            getAssetsWithCache(assetIdsChunk, assetCache$, totalBalance$, assetProvider, maxAssetInfoCacheAge),
           retryBackoffConfig,
+          sample: () =>
+            getAssetsWithCache(assetIdsChunk, assetCache$, totalBalance$, assetProvider, maxAssetInfoCacheAge),
           trigger$: of(true) // fetch only once
         })
       )