I tinkered with an alternative to stream interfaces:

allows you to do things like

    infiniteRecords
    .map(item => doSomeAsyncThing(item), { concurrency: 5 });

And then because I found that I often want to switch between batching items vs dealing with single items:

    infiniteRecords
    .map(item => doSomeAsyncSingularThing(item), { concurrency: 5 })
    .map(groupOf10 => doSomeBatchThing(groupsOf10), { batchSize: 10 })
    // Can flatten back to single items
    .map(item => backToSingleItem(item), { flatten: true });