PubSub3: SwaapV2 & Bebop & Cables

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@paraswap/dex-lib",
-  "version": "4.0.6-rfq-pub-sub.0",
+  "version": "4.0.6-rfq-pub-sub.16",
   "main": "build/index.js",
   "types": "build/index.d.ts",
   "repository": "https://github.com/paraswap/paraswap-dex-lib",

src/dex/bebop/bebop.ts

@@ -73,7 +73,6 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
   private tokensMap: TokenDataMap = {};
 
   private pricesCacheKey: string;
-  private tokensCacheKey: string;
   private tokensAddrCacheKey: string;
 
   private bebopAuthToken: string;
@@ -90,7 +89,6 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
   ) {
     super(dexHelper, dexKey);
     this.logger = dexHelper.getLogger(dexKey);
-    this.tokensCacheKey = `tokens`;
     this.pricesCacheKey = `prices`;
     this.tokensAddrCacheKey = `tokens_addr`;
     const token = this.dexHelper.config.data.bebopAuthToken;
@@ -109,7 +107,6 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
           tokensIntervalMs: BEBOP_TOKENS_POLLING_INTERVAL_MS,
           pricesCacheKey: this.pricesCacheKey,
           pricesCacheTTLSecs: BEBOP_PRICES_CACHE_TTL,
-          tokensCacheKey: this.tokensCacheKey,
           tokensAddrCacheKey: this.tokensAddrCacheKey,
           tokensCacheTTLSecs: BEBOP_TOKENS_CACHE_TTL,
           tokensReqParams: {
@@ -132,11 +129,8 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
   }
 
   async initializePricing(blockNumber: number) {
-    if (!this.dexHelper.config.isSlave) {
-      this.rateFetcher.start();
-      await sleep(BEBOP_INIT_TIMEOUT_MS);
-    }
-
+    await this.rateFetcher.start();
+    await sleep(BEBOP_INIT_TIMEOUT_MS);
     await this.setTokensMap();
   }
 
@@ -184,7 +178,7 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
       return [];
     }
 
-    const prices = await this.getCachedPrices();
+    const prices = await this.rateFetcher.getCachedPrices();
     if (!prices) {
       throw new Error('No prices available');
     }
@@ -523,7 +517,7 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
   }
 
   async setTokensMap() {
-    const tokens = await this.getCachedTokens();
+    const tokens = await this.rateFetcher.getCachedTokens();
 
     if (tokens) {
       this.tokensMap = tokens;
@@ -542,7 +536,7 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
     tokenAddress: Address,
     limit: number,
   ): Promise<PoolLiquidity[]> {
-    const prices = await this.getCachedPrices();
+    const prices = await this.rateFetcher.getCachedPrices();
 
     if (!prices) {
       return [];
@@ -812,13 +806,7 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
             errorsData.count + 1
           } within ${BEBOP_RESTRICT_CHECK_INTERVAL_MS / 1000 / 60} minutes`,
         );
-        await this.dexHelper.cache.setex(
-          this.dexKey,
-          this.network,
-          BEBOP_RESTRICTED_CACHE_KEY,
-          BEBOP_RESTRICT_TTL_S,
-          'true',
-        );
+        this.rateFetcher.restrict();
       } else {
         this.logger.warn(
           `${this.dexKey}-${this.network}: Error count increased`,
@@ -839,41 +827,7 @@ export class Bebop extends SimpleExchange implements IDex<BebopData> {
   }
 
   async isRestricted(): Promise<boolean> {
-    const result = await this.dexHelper.cache.get(
-      this.dexKey,
-      this.network,
-      BEBOP_RESTRICTED_CACHE_KEY,
-    );
-
-    return result === 'true';
-  }
-
-  async getCachedPrices(): Promise<BebopPricingResponse | null> {
-    const cachedPrices = await this.dexHelper.cache.get(
-      this.dexKey,
-      this.network,
-      this.pricesCacheKey,
-    );
-
-    if (cachedPrices) {
-      return JSON.parse(cachedPrices) as BebopPricingResponse;
-    }
-
-    return null;
-  }
-
-  async getCachedTokens(): Promise<TokenDataMap | null> {
-    const cachedTokens = await this.dexHelper.cache.get(
-      this.dexKey,
-      this.network,
-      this.tokensAddrCacheKey,
-    );
-
-    if (cachedTokens) {
-      return JSON.parse(cachedTokens) as TokenDataMap;
-    }
-
-    return null;
+    return this.rateFetcher.isRestricted();
   }
 
   getTokenFromAddress(address: Address): Token {

src/dex/bebop/rate-fetcher.ts

@@ -9,14 +9,13 @@ import {
   BebopPricingResponse,
   BebopRateFetcherConfig,
   BebopTokensResponse,
+  TokenDataMap,
 } from './types';
-import {
-  BebopPricingUpdate,
-  pricesResponseValidator,
-  tokensResponseValidator,
-} from './validators';
+import { BebopPricingUpdate, tokensResponseValidator } from './validators';
 import { WebSocketFetcher } from '../../lib/fetcher/wsFetcher';
 import { utils } from 'ethers';
+import { BEBOP_RESTRICT_TTL_S, BEBOP_RESTRICTED_CACHE_KEY } from './constants';
+import { ExpKeyValuePubSub } from '../../lib/pub-sub';
 
 export function levels_from_flat_array(values: number[]): BebopLevel[] {
   const levels: BebopLevel[] = [];
@@ -27,15 +26,18 @@ export function levels_from_flat_array(values: number[]): BebopLevel[] {
 }
 
 export class RateFetcher {
+  private tokensPricesPubSub: ExpKeyValuePubSub;
+
   private pricesFetcher: WebSocketFetcher<BebopPricingResponse>;
   private pricesCacheKey: string;
   private pricesCacheTTL: number;
 
   private tokensFetcher: Fetcher<BebopTokensResponse>;
   private tokensAddrCacheKey: string;
-  private tokensCacheKey: string;
   private tokensCacheTTL: number;
 
+  private restrictPubSub: ExpKeyValuePubSub;
+
   constructor(
     private dexHelper: IDexHelper,
     private dexKey: string,
@@ -45,6 +47,13 @@ export class RateFetcher {
   ) {
     this.pricesCacheKey = config.rateConfig.pricesCacheKey;
     this.pricesCacheTTL = config.rateConfig.pricesCacheTTLSecs;
+
+    this.tokensPricesPubSub = new ExpKeyValuePubSub(
+      this.dexHelper,
+      this.dexKey,
+      'tokensPrices',
+    );
+
     this.pricesFetcher = new WebSocketFetcher<BebopPricingResponse>(
       {
         info: {
@@ -68,8 +77,8 @@ export class RateFetcher {
     );
 
     this.tokensAddrCacheKey = config.rateConfig.tokensAddrCacheKey;
-    this.tokensCacheKey = config.rateConfig.tokensCacheKey;
     this.tokensCacheTTL = config.rateConfig.tokensCacheTTLSecs;
+
     this.tokensFetcher = new Fetcher<BebopTokensResponse>(
       dexHelper.httpRequest,
       {
@@ -87,6 +96,14 @@ export class RateFetcher {
       config.rateConfig.tokensIntervalMs,
       logger,
     );
+
+    this.restrictPubSub = new ExpKeyValuePubSub(
+      dexHelper,
+      dexKey,
+      'restrict',
+      'not_restricted',
+      BEBOP_RESTRICT_TTL_S,
+    );
   }
 
   parsePricingUpdate(updateObject: any): BebopPricingResponse {
@@ -114,8 +131,13 @@ export class RateFetcher {
   }
 
   start() {
-    this.pricesFetcher.startPolling();
-    this.tokensFetcher.startPolling();
+    if (!this.dexHelper.config.isSlave) {
+      this.pricesFetcher.startPolling();
+      this.tokensFetcher.startPolling();
+    } else {
+      this.tokensPricesPubSub.subscribe();
+      this.restrictPubSub.subscribe();
+    }
   }
 
   stop() {
@@ -141,17 +163,14 @@ export class RateFetcher {
     this.dexHelper.cache.setex(
       this.dexKey,
       this.network,
-      this.tokensCacheKey,
+      this.tokensAddrCacheKey,
       this.tokensCacheTTL,
-      JSON.stringify(tokenMap),
+      JSON.stringify(tokenAddrMap),
     );
 
-    this.dexHelper.cache.setex(
-      this.dexKey,
-      this.network,
-      this.tokensAddrCacheKey,
+    this.tokensPricesPubSub.publish(
+      { [this.tokensAddrCacheKey]: tokenAddrMap },
       this.tokensCacheTTL,
-      JSON.stringify(tokenAddrMap),
     );
   }
 
@@ -179,5 +198,57 @@ export class RateFetcher {
       this.pricesCacheTTL,
       JSON.stringify(normalizedPrices),
     );
+
+    this.tokensPricesPubSub.publish(
+      { [this.pricesCacheKey]: normalizedPrices },
+      this.pricesCacheTTL,
+    );
+  }
+
+  async getCachedPrices(): Promise<BebopPricingResponse | null> {
+    const cachedPrices = await this.tokensPricesPubSub.getAndCache(
+      this.pricesCacheKey,
+    );
+
+    if (cachedPrices) {
+      return cachedPrices as BebopPricingResponse;
+    }
+
+    return null;
+  }
+
+  async getCachedTokens(): Promise<TokenDataMap | null> {
+    const cachedTokens = await this.tokensPricesPubSub.getAndCache(
+      this.tokensAddrCacheKey,
+    );
+
+    if (cachedTokens) {
+      return cachedTokens as TokenDataMap;
+    }
+
+    return null;
+  }
+
+  async isRestricted(): Promise<boolean> {
+    const result = await this.restrictPubSub.getAndCache(
+      BEBOP_RESTRICTED_CACHE_KEY,
+    );
+
+    return result === 'true';
+  }
+
+  async restrict(): Promise<void> {
+    await this.dexHelper.cache.setex(
+      this.dexKey,
+      this.network,
+      BEBOP_RESTRICTED_CACHE_KEY,
+      BEBOP_RESTRICT_TTL_S,
+      'true',
+    );
+
+    this.restrictPubSub.publish(
+      { [BEBOP_RESTRICTED_CACHE_KEY]: 'true' },
+      BEBOP_RESTRICT_TTL_S,
+    );
   }
 }

src/dex/bebop/types.ts

@@ -16,7 +16,6 @@ export type BebopRateFetcherConfig = {
     tokensIntervalMs: number;
     pricesCacheKey: string;
     tokensAddrCacheKey: string;
-    tokensCacheKey: string;
     pricesCacheTTLSecs: number;
     tokensCacheTTLSecs: number;
   };