Added bottleneck to deal with rate-limiting

packages/pipeline/src/data_sources/dex_prices/index.ts

@@ -1,5 +1,7 @@
-import { fetchAsync } from '@0x/utils';
+import { fetchAsync, logUtils } from '@0x/utils';
+import Bottleneck from 'bottleneck';
 
+const ONE_SECOND = 1000;
 const EDPS_BASE_URL = 'http://35.185.219.196:1337';
 
 export type EdpsResponse = EdpsWrapper[];
@@ -21,12 +23,21 @@ export interface EdpsExchange {
 // tslint:disable:prefer-function-over-method
 // ^ Keep consistency with other sources and help logical organization
 export class EdpsSource {
+    // rate-limit for all API calls through this class instance
+    private readonly _limiter: Bottleneck;
+    constructor(maxReqsPerSecond: number) {
+        this._limiter = new Bottleneck({
+            minTime: ONE_SECOND / maxReqsPerSecond,
+        });
+    }
+
     /**
      * Call Ethereum DEX Price Service API.
      */
     public async getEdpsAsync(direction: string, symbol: string, amount: number): Promise<EdpsWrapper> {
         const edpsUrl = `${EDPS_BASE_URL}/${direction}?amount=${amount}&symbol=${symbol}&decimals=`;
-        const resp = await fetchAsync(edpsUrl);
+        logUtils.log(`${direction} ${amount} ${symbol}`);
+        const resp = await this._limiter.schedule(() => fetchAsync(edpsUrl));
         const respJson: EdpsResponse = await resp.json();
         const allExchanges: EdpsWrapper = {};
         // The below unwraps the response so we get 1 single EdpsWrapper object

packages/pipeline/src/scripts/pull_slippage.ts

@@ -12,6 +12,9 @@ import { handleError } from '../utils';
 // Number of orders to save at once.
 const BATCH_SAVE_SIZE = 1000;
 
+// Max requests to make to API per second;
+const EDPS_MAX_REQUESTS_PER_SECOND = 1;
+
 // Maximum requests per second to CryptoCompare
 const CRYPTO_COMPARE_MAX_REQS_PER_SECOND = 60;
 
@@ -26,7 +29,7 @@ let connection: Connection;
 
 (async () => {
     connection = await createConnection(ormConfig as ConnectionOptions);
-    const edpsSource = new EdpsSource();
+    const edpsSource = new EdpsSource(EDPS_MAX_REQUESTS_PER_SECOND);
     const cryptoCompareSource = new CryptoCompareOHLCVSource(CRYPTO_COMPARE_MAX_REQS_PER_SECOND);
 
     logUtils.log('Fetching slippage records');