Как объединить данные из нескольких полезных нагрузок в Node-RED?

Question

Я хочу объединить данные (из HTTP msg.payload) из 3 разных источников.

Однако эти HTTP-запросы могут вызываться несколько раз, поэтому данные из одного и того же источника могут приниматься более одного раза.

[{"id":"7ed13b41.131b14","type":"join","z":"246eac57.42ec74","name":"","mode":"auto","build":"string","property":"payload","propertyType":"msg","key":"index","joiner":"","joinerType":"str","accumulate":false,"timeout":"","count":"3","reduceRight":false,"reduceExp":"","reduceInit":"","reduceInitType":"","reduceFixup":"","x":1370,"y":1160,"wires":[["d941ca6e.0e1aa8"]]},{"id":"d941ca6e.0e1aa8","type":"debug","z":"246eac57.42ec74","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","x":1490,"y":1120,"wires":[]},{"id":"b04a312.d6c40d","type":"function","z":"246eac57.42ec74","name":"part 1","func":"msg.parts = {};\nmsg.parts.id = 12345;\nmsg.parts.index = 0;\nmsg.parts.count = 3;\nreturn msg;","outputs":1,"noerr":0,"x":1210,"y":1120,"wires":[["7ed13b41.131b14"]]},{"id":"30cec12.e2fc13e","type":"function","z":"246eac57.42ec74","name":"part 2","func":"msg.parts = {};\nmsg.parts.id = 12345;\nmsg.parts.index = 1;\nmsg.parts.count = 3;\nreturn msg;","outputs":1,"noerr":0,"x":1210,"y":1160,"wires":[["7ed13b41.131b14"]]},{"id":"8902f2d5.ea688","type":"function","z":"246eac57.42ec74","name":"part 3","func":"msg.parts = {};\nmsg.parts.id = 12345;\nmsg.parts.index = 2;\nmsg.parts.count = 3;\nreturn msg;","outputs":1,"noerr":0,"x":1210,"y":1200,"wires":[["7ed13b41.131b14"]]},{"id":"814f25b6.dd3958","type":"http in","z":"246eac57.42ec74","name":"source 1","url":"/source1","method":"get","upload":false,"swaggerDoc":"","x":1060,"y":1120,"wires":[["b04a312.d6c40d"]]},{"id":"cab634ac.5d9df8","type":"http in","z":"246eac57.42ec74","name":"source 2","url":"/source 2","method":"get","upload":false,"swaggerDoc":"","x":1060,"y":1160,"wires":[["30cec12.e2fc13e"]]},{"id":"98f89b04.9b5bb8","type":"http in","z":"246eac57.42ec74","name":"source 3","url":"/source3","method":"get","upload":false,"swaggerDoc":"","x":1060,"y":1200,"wires":[["8902f2d5.ea688"]]}]

Что происходит в этом потоке, когда узел присоединения получает 3 сообщения от источника 1, он считает msg.parts завершенным. Поведение, которого я хочу достичь, заключается в том, что только когда данные из 3 источников будут получены, поток будет продолжаться. И если данные получены из того же источника, они будут перезаписывать только предыдущие данные.

Есть ли способ достичь этого в Node-RED?

Answer 1

Один из способов - использовать разные msg.topic для каждого источника и настроить узел join, чтобы вручную объединить все входные данные в один объект, связанный с темой:

[{"id":"7f9b1fbb.58ff","type":"join","z":"f9a2eec9.c2e26","name":"","mode":"custom","build":"object","property":"payload","propertyType":"msg","key":"topic","joiner":"","joinerType":"str","accumulate":true,"timeout":"","count":"3","reduceRight":false,"reduceExp":"","reduceInit":"","reduceInitType":"","reduceFixup":"","x":770,"y":2040,"wires":[["42ac1f9e.eac62"]]},{"id":"42ac1f9e.eac62","type":"debug","z":"f9a2eec9.c2e26","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","x":930,"y":2040,"wires":[]},{"id":"37be742c.72a96c","type":"http in","z":"f9a2eec9.c2e26","name":"source 1","url":"/source1/:value","method":"get","upload":false,"swaggerDoc":"","x":180,"y":2020,"wires":[["8ca164f9.828898"]]},{"id":"51684d03.091ea4","type":"http in","z":"f9a2eec9.c2e26","name":"source 2","url":"/source2/:value","method":"get","upload":false,"swaggerDoc":"","x":180,"y":2060,"wires":[["2bb81dc1.351562"]]},{"id":"1a7ad806.dbb598","type":"http in","z":"f9a2eec9.c2e26","name":"source 3","url":"/source3/:value","method":"get","upload":false,"swaggerDoc":"","x":180,"y":2100,"wires":[["c7da5f3a.95b71"]]},{"id":"9a85c694.2a37a8","type":"http response","z":"f9a2eec9.c2e26","name":"","statusCode":"204","headers":{},"x":780,"y":2080,"wires":[]},{"id":"2a0579a6.612c66","type":"change","z":"f9a2eec9.c2e26","name":"set payload value","rules":[{"t":"set","p":"payload","pt":"msg","to":"req.params.value","tot":"msg"}],"action":"","property":"","from":"","to":"","reg":false,"x":570,"y":2060,"wires":[["9a85c694.2a37a8","7f9b1fbb.58ff"]]},{"id":"8ca164f9.828898","type":"change","z":"f9a2eec9.c2e26","name":"set topic 1","rules":[{"t":"set","p":"topic","pt":"msg","to":"source1","tot":"str"}],"action":"","property":"","from":"","to":"","reg":false,"x":350,"y":2020,"wires":[["2a0579a6.612c66"]]},{"id":"2bb81dc1.351562","type":"change","z":"f9a2eec9.c2e26","name":"set topic 2","rules":[{"t":"set","p":"topic","pt":"msg","to":"source2","tot":"str"}],"action":"","property":"","from":"","to":"","reg":false,"x":350,"y":2060,"wires":[["2a0579a6.612c66"]]},{"id":"c7da5f3a.95b71","type":"change","z":"f9a2eec9.c2e26","name":"set topic 3","rules":[{"t":"set","p":"topic","pt":"msg","to":"source3","tot":"str"}],"action":"","property":"","from":"","to":"","reg":false,"x":350,"y":2100,"wires":[["2a0579a6.612c66"]]}]

Как уже упоминалось в комментариях, вам нужно вернуть что-то из всех ваших потоков, которые начинаются с http in узлов ... обычно я люблю использовать конечные точки http POST (для получения входных значений) и вернуть код состояния 204 (без содержимого), чтобы сообщить вызывающему абоненту, что значение было принято. Поскольку ваш исходный поток использует узлы GET, я параметризировал URL-адреса (например, /source1/:value) и использовал узел change, чтобы скопировать значение в полезную нагрузку. Ваш пробег может отличаться ...

Обратите внимание, что я установил для "количества частей сообщения" значение 3, что означает, что вы не получите никакого вывода от узла join, пока он не получит 3 различных значений темы. После этого каждое входное сообщение выводит последний объединенный объект сообщения (из-за флажка «каждое последующее сообщение»). Изменяя эти два варианта, вы можете получить любое поведение, которое вам нужно.