Node-RED: split-joinのメモリ増加を減らす方法

便利なsplit-join

splitノードとjoinノードの組み合わせは、Node-REDで複数データを非同期処理する場合に便利な方法です。splitノードは配列など複数データをバラバラにし、何らかの処理の後、バラバラのデータを元の形式にまとめてくれます。

暗黙のクローンを作るsplit-join

しかし、以前の記事で書いたように、payload以外のmsgのデータがディープコピーされたクローンを作ってしまいます。少量のデータであれば問題ないかもしれませんが、非同期処理の対象データ数が多い場合は、非同期処理に不要なデータまで大量に複製されてしまいます。
そこで、メモリをあまり使わない方法を2つ考えましたので公開します。

安全な方法

安全な方法はsplit-joinを2重にします。外側（最初）のsplit-joinは非同期処理の対象データとそれ以外のデータを分ける目的で、内側（2段目）のsplit-joinはもともとやりたかったsplit-joinです。
以下のような処理になります

1. msg.payloadの配列の最初の元々のpayloadを入れ、2番目以降（1つでも2つでも良い）に元々のpayload以外の属性を入れる
2. 外側のsplit
3. 　msg.parts.indexが0以外なら、何もせず外側のjoinへ
4. 　内側（本来の）のsplit
5. 　　対象データの非同期処理
6. 　内側（本来の）のjoin
7. 外側のjoin
8. 後続処理
   この方法では外側のsplitによって１度だけcloneが作られてしまいますが、例えば非同期処理したいデータ数が3000だとするとcloneの作成回数、つまりメモリ消費は1/3000になります。
   多くの場合、この方法で十分でしょう。

要注意な方法

１度のコピーもしたくない場合を考えて、cloneを作らないでshllow copyするsplitをfunctionノードで実装してみました。deep copyでなくshllow copyですから、msgのpayload以外の属性を変更すると全体に影響しますので気を付けてください。配列のみ対応しています。ご自由に使っていただいて構いませんが無保証です。

node.js

if (!Array.isArray(msg.payload)) {
    return msg;
}
let parts = msg.parts;
let payload = msg.payload;
let id = RED.util.generateId();    // generate a random id
let count = payload.length;

payload.forEach((item, ix) => {
    // Shallow copy: using Object.assign to copy top-level properties
    let newmsg = Object.assign({}, msg);
    newmsg.payload = item;
    newmsg.parts = {};
    if (msg.hasOwnProperty("parts")) {
        newmsg.parts.parts = parts;
    }
    newmsg.parts.id = id;
    newmsg.parts.type = "array";
    newmsg.parts.count = count;
    newmsg.parts.index = ix;
    newmsg.parts.len = 1;

    node.send(newmsg, false);
});
return;

ポイントはmsg.partsです。splitが多段になるとmsg.partsにsplit前のmsg.partsを保存することで、正しくjoinできます。

サンプルフロー

[{"id":"5850bb17ae3a27cc","type":"function","z":"c84a4fc3759a0cd9","name":"shallow split","func":"if (!Array.isArray(msg.payload)) {\n    return msg;\n}\nlet parts = msg.parts;\nlet payload = msg.payload;\nlet id = RED.util.generateId();    // generate a random id\nlet count = payload.length;\n\npayload.forEach((item, ix) => {\n    // Shallow copy: using Object.assign to copy top-level properties\n    let newmsg = Object.assign({}, msg);\n    newmsg.payload = item;\n    newmsg.parts = {};\n    if (msg.hasOwnProperty(\"parts\")) {\n        newmsg.parts.parts = parts;\n    }\n    newmsg.parts.id = id;\n    newmsg.parts.type = \"array\";\n    newmsg.parts.count = count;\n    newmsg.parts.index = ix;\n    newmsg.parts.len = 1;\n\n    node.send(newmsg, false);\n});\nreturn;","outputs":1,"timeout":0,"noerr":0,"initialize":"","finalize":"","libs":[],"x":270,"y":740,"wires":[["179c846005acf52e"]]},{"id":"7bc26da08cecc657","type":"inject","z":"c84a4fc3759a0cd9","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"[[0,1,2],[3,4,5]]","payloadType":"json","x":110,"y":720,"wires":[["5850bb17ae3a27cc"]]},{"id":"179c846005acf52e","type":"function","z":"c84a4fc3759a0cd9","name":"shallow split","func":"if (!Array.isArray(msg.payload)) {\n    return msg;\n}\nlet parts = msg.parts;\nlet payload = msg.payload;\nlet id = RED.util.generateId();    // generate a random id\nlet count = payload.length;\n\npayload.forEach((item, ix) => {\n    // Shallow copy: using Object.assign to copy top-level properties\n    let newmsg = Object.assign({}, msg);\n    newmsg.payload = item;\n    newmsg.parts = {};\n    if (msg.hasOwnProperty(\"parts\")) {\n        newmsg.parts.parts = parts;\n    }\n    newmsg.parts.id = id;\n    newmsg.parts.type = \"array\";\n    newmsg.parts.count = count;\n    newmsg.parts.index = ix;\n    newmsg.parts.len = 1;\n\n    node.send(newmsg, false);\n});\nreturn;","outputs":1,"timeout":0,"noerr":0,"initialize":"","finalize":"","libs":[],"x":430,"y":760,"wires":[["e1af61b73a252d4c"]]},{"id":"68c618ae601173f0","type":"join","z":"c84a4fc3759a0cd9","name":"","mode":"auto","build":"object","property":"payload","propertyType":"msg","key":"topic","joiner":"\\n","joinerType":"str","useparts":false,"accumulate":"false","timeout":"","count":"","reduceRight":false,"x":410,"y":820,"wires":[["d8caec8deaccbc6f"]]},{"id":"d8caec8deaccbc6f","type":"join","z":"c84a4fc3759a0cd9","name":"","mode":"auto","build":"object","property":"payload","propertyType":"msg","key":"topic","joiner":"\\n","joinerType":"str","useparts":false,"accumulate":"false","timeout":"","count":"","reduceRight":false,"x":250,"y":860,"wires":[["c7c937b6eae1b3fd"]]},{"id":"e1af61b73a252d4c","type":"function","z":"c84a4fc3759a0cd9","name":"msg.payload++","func":"msg.payload++;\nreturn msg;","outputs":1,"timeout":0,"noerr":0,"initialize":"","finalize":"","libs":[],"x":600,"y":780,"wires":[["68c618ae601173f0"]]},{"id":"c7c937b6eae1b3fd","type":"debug","z":"c84a4fc3759a0cd9","name":"結果出力","active":true,"tosidebar":true,"console":false,"tostatus":true,"complete":"payload","targetType":"msg","statusVal":"payload","statusType":"auto","x":580,"y":860,"wires":[]}]