From ad42c98e5a6a5cbf0c95b37b896fff920a3ee723 Mon Sep 17 00:00:00 2001
From: Nicolas Thouvenin <nthouvenin@gmail.com>
Date: Sun, 26 Jul 2020 08:40:32 +0200
Subject: [PATCH] =?UTF-8?q?fix(analytics):=20=F0=9F=90=9B=20attemps=20to?=
 =?UTF-8?q?=20fix=20combine=20too?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 packages/analytics/src/combine.js  | 27 +++++++++++++++++----------
 packages/analytics/test/combine.js |  1 +
 2 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/packages/analytics/src/combine.js b/packages/analytics/src/combine.js
index 380357d2..fda757cb 100644
--- 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
index 37dba3fd..a469a746 100644
--- 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) => {