page-stream.js

const {Readable} = require('stream');

class PageStream extends Readable {
  constructor(requestFn) {
    super({objectMode: true}); 
    this.request = requestFn;
    this.nextQuery = {};
    this.pending = [];
  }
  async _read() {
    if (!this.pending.length) {
      if (!this.nextQuery) {
        this.push(null);
        return;
      }
      
      const [results, nextQuery] = await this.request(this.nextQuery);
      this.pending = results;
      this.nextQuery = nextQuery;
    }
    
    let more = true;
    
    while (more && this.pending.length) {
      more = this.push(this.pending.shift());
    }
    
    if (more) {
       setImmediate(() => this._read()); 
    }
  }
}