Merge pull request #450 from oceanprotocol/feature/chunk_promises

chunk promises

src/balancer/OceanPool.ts

@@ -11,6 +11,7 @@ declare type PoolTransactionType = 'swap' | 'join' | 'exit'
 const POOL_MAX_AMOUNT_IN_LIMIT = 0.25 // maximum 1/4 of the pool reserve
 const POOL_MAX_AMOUNT_OUT_LIMIT = 0.25 // maximum 1/4 of the pool reserve
 const BPFACTORY_DEPLOY_BLOCK = 0
+const MAX_AWAIT_PROMISES = 10 // infura has a limit of 10 requests/sec
 export interface PoolDetails {
   poolAddress: string
   tokens: string[]
@@ -957,24 +958,34 @@ export class OceanPool extends Pool {
   public async getPoolSharesByAddress(account: string): Promise<PoolShare[]> {
     const result: PoolShare[] = []
     const factory = new this.web3.eth.Contract(this.factoryABI, this.factoryAddress)
-
     const events = await factory.getPastEvents('BPoolRegistered', {
       filter: {},
       fromBlock: this.startBlock,
       toBlock: 'latest'
     })
-    let results = await Promise.all(
+    let promises = []
-      events.map((event) => {
+    for (let i = 0; i < events.length; i++) {
-        return this.getResult(account, event)
+      promises.push(this.getResult(account, events[i]))
-      })
+      if (promises.length > MAX_AWAIT_PROMISES) {
-    )
+        const results = await Promise.all(promises)
-    results = results.filter((share) => {
+        for (let j = 0; j < results.length; j++) {
+          result.push(results[j])
+        }
+        promises = []
+      }
+    }
+    if (promises.length > 0) {
+      const results = await Promise.all(promises)
+      for (let j = 0; j < results.length; j++) {
+        result.push(results[j])
+      }
+      promises = []
+    }
+
+    const filteredResult = result.filter((share) => {
       return share !== undefined
     })
-
+    return filteredResult
-    result.push(...results)
-
-    return result
   }
 
   /**
@@ -1016,25 +1027,24 @@ export class OceanPool extends Pool {
       toBlock: 'latest'
     })
 
-    let eventResults = await Promise.all(
+    let promises = []
-      events.map((event) => {
+    for (let i = 0; i < events.length; i++) {
-        switch (event.topics[0]) {
+      promises.push(this.getEventData(poolAddress, dtAddress, events[i]))
-          case swapTopic:
+      if (promises.length > MAX_AWAIT_PROMISES) {
-            return this.getEventData('swap', poolAddress, dtAddress, event)
+        const data = await Promise.all(promises)
-            break
+        for (let j = 0; j < data.length; j++) {
-          case joinTopic:
+          results.push(data[j])
-            return this.getEventData('join', poolAddress, dtAddress, event)
+        }
-            break
+        promises = []
-          case exitTopic:
+      }
-            return this.getEventData('exit', poolAddress, dtAddress, event)
+    }
-            break
+    if (promises.length > 0) {
+      const data = await Promise.all(promises)
+      for (let j = 0; j < data.length; j++) {
+        results.push(data[j])
+      }
+      promises = []
     }
-      })
-    )
-    eventResults = eventResults.filter((share) => {
-      return share !== undefined
-    })
-    results.push(...eventResults)
 
     // for (let i = 0; i < events.length; i++) {
     //   switch (events[i].topics[0]) {
@@ -1049,7 +1059,10 @@ export class OceanPool extends Pool {
     //       break
     //   }
     // }
-    return results
+    const eventResults = results.filter((share) => {
+      return share !== undefined
+    })
+    return eventResults
   }
 
   /**
@@ -1058,7 +1071,7 @@ export class OceanPool extends Pool {
    * @return {PoolTransaction[]}
    */
   public async getAllPoolLogs(account: string): Promise<PoolTransaction[]> {
-    const results: PoolTransaction[] = []
+    const results: PoolTransaction[][] = []
     const factory = new this.web3.eth.Contract(this.factoryABI, this.factoryAddress)
     const events = await factory.getPastEvents('BPoolRegistered', {
       filter: {},
@@ -1066,25 +1079,53 @@ export class OceanPool extends Pool {
       toBlock: 'latest'
     })
 
-    const eventsResults = await Promise.all(
+    let promises = []
-      events.map((event) => {
+    for (let i = 0; i < events.length; i++) {
-        return this.getPoolLogs(event.returnValues[0], event.blockNumber, account)
+      promises.push(
-      })
+        this.getPoolLogs(events[i].returnValues[0], events[i].blockNumber, account)
       )
+      if (promises.length > MAX_AWAIT_PROMISES) {
+        const data = await Promise.all(promises)
+        for (let j = 0; j < data.length; j++) {
+          results.push(data[j])
+        }
+        promises = []
+      }
+    }
+    if (promises.length > 0) {
+      const data = await Promise.all(promises)
+      for (let j = 0; j < data.length; j++) {
+        results.push(data[j])
+      }
+      promises = []
+    }
 
-    const eventResults = eventsResults.reduce((elem1, elem2) => elem1.concat(elem2))
+    const concatResults = results.reduce((elem1, elem2) => elem1.concat(elem2))
-    results.push(...eventResults)
 
-    return results
+    return concatResults
   }
 
   private async getEventData(
-    type: PoolTransactionType,
     poolAddress: string,
     dtAddress: string,
     data: Log
   ): Promise<PoolTransaction> {
     const blockDetails = await this.web3.eth.getBlock(data.blockNumber)
+    const swapTopic = super.getSwapEventSignature()
+    const joinTopic = super.getJoinEventSignature()
+    const exitTopic = super.getExitEventSignature()
+    let type: PoolTransactionType
+    switch (data.topics[0]) {
+      case swapTopic:
+        type = 'swap'
+        break
+      case joinTopic:
+        type = 'join'
+        break
+      case exitTopic:
+        type = 'exit'
+        break
+    }
     let result: PoolTransaction = {
       poolAddress,
       dtAddress,