cache many small periods on disk and aggregate them

1 files changed, 38 insertions, 0 deletions

diff --git a/source.js b/source.js
new file mode 100644
index 0000000..2d02939
--- /dev/null
+++ b/source.js
@@ -0,0 +1,38 @@
+#!/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))
+    .chain(/*R.map(R.chain(Future.after(intervalSleep),*/ saveFPayloads(timestamp))/*))*/
+    .chain(Future.after(intervalSleep));
+  // TODO sleeps 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,
+};