scramjet.js

const scramjet = require("scramjet");

const {getAPI, insertDB} = require('./io-simulators')
const ROWS = 1000

// build a stream of indexes
new scramjet.DataStream({
     read() {
       this.offset = (this.offset || 0);
       this.push(this.offset);
       this.offset += ROWS ;
     }
  })
  // map indexes to API responses
  .map(i => getAPI(i, ROWS))
  .flatMap((i) => i)
  // filter out the id's containing "3"
  .filter(function(item) {
    return (""+item.id).indexOf('3') > -1
  })
  // add timestamp to id's containing "9"
  .map(function(item) {
    if ((""+item.id).indexOf('9') > -1) {
      return Object.assign(item, {timestamp: Date.now()})
    } else {
      return item
    }
  })
  // write to database
· .accumulate(
    function(insert, item) {
      return insert(item)
    },
    insertDB
  );