From dedf7a4fee3f229449e499be8d84f5ff68bd1262 Mon Sep 17 00:00:00 2001 From: Harvey Zuccon Date: Tue, 20 May 2025 02:05:08 +0000 Subject: [PATCH] Resolve "feat(Rates): Composite Adapter should be able to add/remove adapters" --- packages/stack/src/rates/rates-composite.ts | 130 +++++++++++++++----- 1 file changed, 97 insertions(+), 33 deletions(-) diff --git a/packages/stack/src/rates/rates-composite.ts b/packages/stack/src/rates/rates-composite.ts index 58b6f90..e39f9c0 100644 --- a/packages/stack/src/rates/rates-composite.ts +++ b/packages/stack/src/rates/rates-composite.ts @@ -1,28 +1,32 @@ import { BaseRates } from './base-rates.js'; +import { UUIDV7 } from '../utils/uuid.js'; /** * A composite rate adapter that combines rates from multiple rate sources * using a configurable policy. */ export class RatesComposite extends BaseRates { + // List of the adapters with a unique ID for each adapter + private adapters: { id: string; adapter: BaseRates }[] = []; + // Track available pairs across all rate adapters private availablePairs: Set = new Set(); // Cache of the latest rates from each adapter - private rateCache: Map>> = new Map(); + private rateCache: Map>> = new Map(); // Off callbacks for unsubscribing from events - private unsubscribeFunctions: (() => void)[] = []; + private unsubscribeFunctions: Map void> = new Map(); /** * Creates a new RatesComposite instance * * @param adapters Array of BaseRates adapters to combine * @param policy Policy to determine how to combine rates (defaults to average) - * @param refreshIntervalMs How often to emit combined rates (defaults to 1000ms) + * @param refreshIntervalMs How often to emit combined rates (defaults to 60000ms) */ constructor( - private adapters: BaseRates[], + adapters: BaseRates[], private policy: RatesCompositePolicy = ratesAveragePolicy, private refreshIntervalMs: number = 60000, ) { @@ -30,34 +34,17 @@ export class RatesComposite extends BaseRates { // Initialize rate cache this.rateCache = new Map(); + + // Add all adapters to the composite + adapters.forEach((adapter) => this.addAdapter(adapter, false)); } /** * Start all rate adapters and setup event listeners */ public async start(): Promise { - // Setup event listeners for all adapters - for (let i = 0; i < this.adapters.length; i++) { - const adapter = this.adapters[i]; - const adapterIndex = i; - - const listener = (event: { - denominatorUnitCode: string; - numeratorUnitCode: string; - price: number; - }) => { - this.handleRateUpdate(adapterIndex, event); - }; - - // Use the returned unsubscribe function from the custom EventEmitter - const unsubscribe = adapter.on('rateUpdated', listener); - - // Store unsubscribe function for cleanup - this.unsubscribeFunctions.push(unsubscribe); - } - // Start all adapters - await Promise.all(this.adapters.map((adapter) => adapter.start())); + await Promise.all(this.adapters.map(({ adapter }) => adapter.start())); // Start periodic refresh to emit combined rates this.startPeriodicRefresh(); @@ -74,13 +61,86 @@ export class RatesComposite extends BaseRates { } // Clean up all event listeners by calling their unsubscribe functions - for (const unsubscribe of this.unsubscribeFunctions) { + for (const unsubscribe of this.unsubscribeFunctions.values()) { unsubscribe(); } - this.unsubscribeFunctions = []; + this.unsubscribeFunctions.clear(); // Stop all adapters - await Promise.all(this.adapters.map((adapter) => adapter.stop())); + await Promise.all(this.adapters.map(({ adapter }) => adapter.stop())); + } + + /** + * Adds a new rate adapter to the composite + */ + public addAdapter(adapter: BaseRates, start: boolean = true): void { + // Generate a random ID for the adapter + const adapterId = UUIDV7.generateRandom().toString(); + + // Add the adapter to the list + this.adapters.push({ id: adapterId, adapter }); + + const listener = (event: { + denominatorUnitCode: string; + numeratorUnitCode: string; + price: number; + }) => { + this.handleRateUpdate(adapterId, event); + }; + + // Use the returned unsubscribe function from the custom EventEmitter + const unsubscribe = adapter.on('rateUpdated', listener); + + // Store unsubscribe function associated with the adapter ID + this.unsubscribeFunctions.set(adapterId, unsubscribe); + + // Start the adapter if requested + if (start) { + adapter.start(); + } + } + + /** + * Removes a rate adapter from the composite + */ + public removeAdapter(adapter: BaseRates, stop: boolean = true): void { + // Get the index of the adapter + const adapterId = this.adapters.find( + ({ adapter: storedAdapter }) => storedAdapter === adapter, + )?.id; + + // If the adapter is not found, do nothing + if (!adapterId) { + return; + } + + // Call the unsubscribe function for the adapter + this.unsubscribeFunctions.get(adapterId)?.(); + + // Remove the adapter's rates from the cache + this.rateCache.forEach((numeratorMap) => { + numeratorMap.forEach((rateMap) => { + rateMap.delete(adapterId); + }); + }); + + // Remove the unsubscribe function for the adapter + this.unsubscribeFunctions.delete(adapterId); + + // Remove the adapter from the list + this.adapters = this.adapters.filter(({ id }) => id !== adapterId); + + // Stop the adapter if requested + if (stop) { + adapter.stop(); + } + } + + /** + * Sets the policy for how to combine rates + */ + public setPolicy(policy: RatesCompositePolicy): void { + this.policy = policy; } /** @@ -88,7 +148,9 @@ export class RatesComposite extends BaseRates { */ public async listPairs(): Promise> { // Get available pairs from all adapters - const pairPromises = this.adapters.map((adapter) => adapter.listPairs()); + const pairPromises = this.adapters.map(({ adapter }) => + adapter.listPairs(), + ); const pairSets = await Promise.all(pairPromises); // Combine all pairs @@ -127,7 +189,7 @@ export class RatesComposite extends BaseRates { * Handles a rate update from an adapter */ private handleRateUpdate( - adapterIndex: number, + adapterId: string, event: { denominatorUnitCode: string; numeratorUnitCode: string; @@ -148,7 +210,7 @@ export class RatesComposite extends BaseRates { // Store the rate in the cache const rateMap = denomMap.get(numeratorUnitCode)!; - rateMap.set(adapterIndex, price); + rateMap.set(adapterId, price); // Add pair to available pairs this.availablePairs.add(`${numeratorUnitCode}/${denominatorUnitCode}`); @@ -162,8 +224,10 @@ export class RatesComposite extends BaseRates { for (const [denominator, numeratorMap] of this.rateCache.entries()) { // Process each numerator currency for (const [numerator, rateMap] of numeratorMap.entries()) { - // Get all rates for this pair - const rates = Array.from(rateMap.values()); + // Go through our adapters in order, plucking the rate for each adapter + const rates = this.adapters + .map(({ id }) => rateMap.get(id)) + .filter((rate) => rate !== undefined); // Apply policy to get combined rate const combinedRate = this.policy(rates);