From 8007bea7db71e12da1d51ea6fc319a697c2fa0e8 Mon Sep 17 00:00:00 2001
From: Kunihiko Toumura <kunihiko.toumura.yv@hitachi.com>
Date: Mon, 26 Oct 2020 20:25:52 +0900
Subject: [PATCH] Messaging API support in CSV node

---
 .../@node-red/nodes/core/parsers/70-CSV.js    | 21 +++++-----
 test/nodes/core/parsers/70-CSV_spec.js        | 39 +++++++++++++++++++
 2 files changed, 51 insertions(+), 9 deletions(-)

diff --git a/packages/node_modules/@node-red/nodes/core/parsers/70-CSV.js b/packages/node_modules/@node-red/nodes/core/parsers/70-CSV.js
index 15c16da13..b2fe9a4b5 100644
--- a/packages/node_modules/@node-red/nodes/core/parsers/70-CSV.js
+++ b/packages/node_modules/@node-red/nodes/core/parsers/70-CSV.js
@@ -55,7 +55,7 @@ module.exports = function(RED) {
         node.template = clean(node.template);
         node.hdrSent = false;
 
-        this.on("input", function(msg) {
+        this.on("input", function(msg, send, done) {
             if (msg.hasOwnProperty("reset")) {
                 node.hdrSent = false;
             }
@@ -140,9 +140,10 @@ module.exports = function(RED) {
                         }
                         msg.payload = ou;
                         msg.columns = node.template.join(',');
-                        if (msg.payload !== '') { node.send(msg); }
+                        if (msg.payload !== '') { send(msg); }
+                        done();
                     }
-                    catch(e) { node.error(e,msg); }
+                    catch(e) { done(e); }
                 }
                 else if (typeof msg.payload == "string") { // convert CSV string to object
                     try {
@@ -251,13 +252,13 @@ module.exports = function(RED) {
                                     msg.payload = node.store;
                                     msg.columns = node.template.filter(val => val).join(',');
                                     delete msg.parts;
-                                    node.send(msg);
+                                    send(msg);
                                     node.store = [];
                                 }
                             }
                             else {
                                 msg.columns = node.template.filter(val => val).join(',');
-                                node.send(msg); // finally send the array
+                                send(msg); // finally send the array
                             }
                         }
                         else {
@@ -281,19 +282,21 @@ module.exports = function(RED) {
                                         newMessage.parts.count -= 1;
                                     }
                                 }
-                                node.send(newMessage);
+                                send(newMessage);
                             }
                         }
                         node.linecount = 0;
+                        done();
                     }
-                    catch(e) { node.error(e,msg); }
+                    catch(e) { done(e); }
                 }
-                else { node.warn(RED._("csv.errors.csv_js")); }
+                else { node.warn(RED._("csv.errors.csv_js")); done(); }
             }
             else {
                 if (!msg.hasOwnProperty("reset")) {
-                    node.send(msg); // If no payload and not reset - just pass it on.
+                    send(msg); // If no payload and not reset - just pass it on.
                 }
+                done();
             }
         });
     }
diff --git a/test/nodes/core/parsers/70-CSV_spec.js b/test/nodes/core/parsers/70-CSV_spec.js
index 10e167b6d..656f9faa6 100644
--- a/test/nodes/core/parsers/70-CSV_spec.js
+++ b/test/nodes/core/parsers/70-CSV_spec.js
@@ -719,4 +719,43 @@ describe('CSV node', function() {
         });
     });
 
+    it('should call done when message processing is completed', function(done) {
+        const completeNode = require("nr-test-utils").require("@node-red/nodes/core/common/24-complete.js");
+        const flow = [ { id:"n1", type:"csv", temp:"a,b,c,d", wires:[[]]},
+                       { id:"c1", type:"complete", scope: ["n1"], uncaught:false, wires:[["h1"]]},
+                       { id:"h1", type:"helper", wires:[[]]} ];
+        helper.load([csvNode,completeNode], flow, function() {
+            const n1 = helper.getNode("n1");
+            const h1 = helper.getNode("h1");
+            h1.on("input", function(msg) {
+                try {
+                    msg.should.have.a.property('payload', "1,2,3,4");
+                    done();
+                } catch (e) {
+                    done(e);
+                }
+            });
+            n1.receive({payload:"1,2,3,4"});
+        });
+    });
+
+    it('should call done when input causes an error', function(done) {
+        const completeNode = require("nr-test-utils").require("@node-red/nodes/core/common/24-complete.js");
+        const flow = [ { id:"n1", type:"csv", temp:"a,b,c,d", wires:[[]]},
+                       { id:"c1", type:"complete", scope: ["n1"], uncaught:false, wires:[["h1"]]},
+                       { id:"h1", type:"helper", wires:[[]]} ];
+        helper.load([csvNode,completeNode], flow, function() {
+            const n1 = helper.getNode("n1");
+            const h1 = helper.getNode("h1");
+            h1.on("input", function(msg) {
+                try {
+                    msg.should.have.a.property('payload', 1);
+                    done();
+                } catch (e) {
+                    done(e);
+                }
+            });
+            n1.receive({payload:1}); // neither object nor string
+        });
+    });
 });