From 5caf4e48174533710243e8eeaf52567795ec9d01 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 7 May 2018 09:06:04 +0200 Subject: move files to backend subdir, push data in submodule --- backend/source.js | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 backend/source.js (limited to 'backend/source.js') diff --git a/backend/source.js b/backend/source.js new file mode 100644 index 0000000..0f4fd28 --- /dev/null +++ b/backend/source.js @@ -0,0 +1,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, +}; -- cgit v1.3.1