This repository has been archived by the owner on Jul 31, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 41
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add requestUtil.bufferedPut to limit concurrency of put
Deprecate SEND_SYNC_RECORDS category to be optional and derived from each record. Required for #112
- Loading branch information
Showing
9 changed files
with
143 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| 'use strict' | ||
|
|
||
| /** | ||
| * Wrap a Promise-returning function so calls to it fill a queue which has | ||
| * a concurrency limit. | ||
| * e.g. there is an API rate limited to 10 concurrent connections. | ||
| * const getApi = (arg) => window.fetch(arg) | ||
| * const throttledGetApi = limitConcurrency(getApi, 10) | ||
| * for (let i; i < 1000; i++) { throttledGetApi(i) } | ||
| * @param fn {function} Function which returns a Promise | ||
| * @param concurrency {number} Maximum pending/concurrent fn calls | ||
| * @returns {function} | ||
| */ | ||
| module.exports.limitConcurrency = function (fn, concurrency) { | ||
| var queue = null | ||
| var active = [] | ||
| const enqueueFnFactory = function (_this, args) { | ||
| return function () { | ||
| const enqueued = fn.apply(_this, args) | ||
| enqueued.then(function () { | ||
| active.splice(active.indexOf(enqueued), 1) | ||
| }) | ||
| active.push(enqueued) | ||
| return { | ||
| enqueued, | ||
| newQueue: Promise.race(active) | ||
| } | ||
| } | ||
| } | ||
| return function () { | ||
| var enqueueFn = enqueueFnFactory(this, arguments) | ||
| if (active.length < concurrency) { | ||
| const promises = enqueueFn() | ||
| queue = promises.newQueue | ||
| return promises.enqueued | ||
| } else { | ||
| const advanceQueue = queue.then(enqueueFn) | ||
| queue = advanceQueue.then(promises => promises.newQueue) | ||
| return advanceQueue.then(promises => promises.enqueued) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| const test = require('tape') | ||
| const promiseHelper = require('../lib/promiseHelper') | ||
|
|
||
| test('promiseHelper', (t) => { | ||
| t.plan(1) | ||
|
|
||
| t.test('limitConcurrency', (t) => { | ||
| t.plan(1) | ||
|
|
||
| t.test('calls the original function the same number of times with correct args', (t) => { | ||
| t.plan(2) | ||
| const EXPECTED_CALL_COUNT = 100 | ||
| let callCount = 0 | ||
| const asyncFun = (i) => new Promise((resolve, reject) => { | ||
| setTimeout(() => { | ||
| callCount += 1 | ||
| resolve(i) | ||
| }, Math.round(10 * Math.random())) | ||
| }) | ||
| const throttedAsyncFun = promiseHelper.limitConcurrency(asyncFun, 3) | ||
| const promises = [] | ||
| let expectedSum = 0 | ||
| for (let i = 0; i < EXPECTED_CALL_COUNT; i++) { | ||
| promises.push(throttedAsyncFun(i)) | ||
| expectedSum += i | ||
| } | ||
| Promise.all(promises).then((results) => { | ||
| const sum = results.reduce((a, b) => a + b) | ||
| t.equal(callCount, EXPECTED_CALL_COUNT) | ||
| t.equal(sum, expectedSum) | ||
| }) | ||
| }) | ||
| }) | ||
| }) |