fix(analytics): 🐛 attemps to fix combine too

Inist-CNRS · Jul 26, 2020 · ad42c98 · ad42c98
1 parent 66750ed
commit ad42c98
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 10 deletions.
diff --git a/packages/analytics/src/combine.js b/packages/analytics/src/combine.js
@@ -3,6 +3,19 @@ import set from 'lodash.set';
 import debug from 'debug';
 import { createStore } from './store';
 
+async function saveIn(data, feed) {
+    if (this.isLast()) {
+        return feed.close();
+    }
+    const store = this.getEnv();
+    const key = get(data, 'id');
+    const isKey = Boolean(key);
+    if (isKey) {
+        await store.put(key, data);
+    }
+    return feed.send(key);
+}
+
 /**
  * Takes an `Object` and substitute a field with the corresponding value found in a external pipeline
  * the internal pipeline must produce a stream of special object (id, value)
@@ -73,17 +86,11 @@ export default function combine(data, feed) {
             statements = ezs.compileCommands(commands, this.getEnv());
         }
         const output = ezs.createPipeline(input, statements)
+            .pipe(ezs(saveIn, null, this.store))
             .pipe(ezs.catch())
-            .on('data', async (item) => {
-                const key = get(item, 'id');
-                const isKey = Boolean(key);
-
-                if (isKey) {
-                    await this.store.put(key, item);
-                }
-            })
+            .on('data', (d) => d)
             .on('error', (e) => feed.stop(e));
-        whenReady = new Promise((resolve) => output.on('end', resolve));
+        whenReady = new Promise((resolve) => output.on('end', () => resolve));
         input.write(primer);
         input.end();
     }
@@ -107,5 +114,5 @@ export default function combine(data, feed) {
             }
             return feed.send(data);
         })
-        .catch(/* istanbul ignore next */(e) => feed.stop(e));
+        .catch((e) => feed.stop(e));
 }
diff --git a/packages/analytics/test/combine.js b/packages/analytics/test/combine.js
@@ -63,6 +63,7 @@ describe('combine', () => {
 
         from(input)
             .pipe(ezs('combine', { path: 'b', file }))
+            .pipe(ezs('debug', { text: 'after' }))
             .pipe(ezs.catch())
             .on('error', done)
             .on('data', (chunk) => {