source.js


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

#!/usr/bin/node

const R  = require('ramda'),
  Future = require('fluture'),
  etl    = require('./etl'),
  params = require('./params'),
  reduce = require('./reduce'),
  file   = require('./file');

function doFetchProcessStoreHour(config) {
  const intervalSleep = 60 / config.api.intervalsPerMinute * 1000;

  const paramsForHourSample = params.paramsForHourSample(config.api);
  const loadFApi            = etl.loadFPayloads(config.etl);
  const saveFPayloads       = file.saveFPayloadsTimestamped(config.file);
  const aggregatePayloads   = reduce.aggregatePayloads(config.reduce);

  const requestsForHour = (timestamp) => R.map(R.apply(loadFApi), paramsForHourSample(timestamp));

  return (timestamp) => Future.parallel(1, requestsForHour(timestamp))
    .map(R.compose(aggregatePayloads, R.unnest))
    .chainRej((err) => { console.error(err); return Future.of({}); }) // fail silently
    .chain(saveFPayloads(timestamp));
    //.chain(Future.after(intervalSleep));
  // TODO should sleep after request; throttling rpm should happen
  // at a higher level though. CPU / IO / AWS requests stall
  // as well
}

function loadFTimestamped(config) {
  // load from file or fall back to API and store
  return (timestamp) =>
    file.loadFPayloads(config)(config.file.pattern(timestamp))
    .or(doFetchProcessStoreHour(config)(timestamp));
};

module.exports = {
  loadFTimestamped,
};