✨ Add MQTT & Trigger Node (#1705)

* ✨ MQTT-Node

* ⚡ Small fix

* ⚡ Error when the publish method faile

* ⚡ Improvements

* ⚡ Improvements

* ⚡ Add Send Input Data parameter

* ⚡ Minor improvements

Co-authored-by: Jan Oberhauser <jan.oberhauser@gmail.com>

packages/nodes-base/credentials/Mqtt.credentials.ts

@@ -3,15 +3,11 @@ import {
 	NodePropertyTypes,
 } from 'n8n-workflow';
 
-
 export class Mqtt implements ICredentialType {
 	name = 'mqtt';
 	displayName = 'MQTT';
 	documentationUrl = 'mqtt';
 	properties = [
-		// The credentials to get from user and save encrypted.
-		// Properties can be defined exactly in the same way
-		// as node properties.
 		{
 			displayName: 'Protocol',
 			name: 'protocol',
@@ -55,5 +51,19 @@ export class Mqtt implements ICredentialType {
 			},
 			default: '',
 		},
+		{
+			displayName: 'Clean Session',
+			name: 'clean',
+			type: 'boolean' as NodePropertyTypes,
+			default: true,
+			description: `Set to false to receive QoS 1 and 2 messages while offline.`,
+		},
+		{
+			displayName: 'Client ID',
+			name: 'clientId',
+			type: 'string' as NodePropertyTypes,
+			default: '',
+			description: 'Client ID. If left empty, one is autogenrated for you',
+		},
 	];
 }

packages/nodes-base/nodes/MQTT/Mqtt.node.json

@@ -0,0 +1,21 @@
+{
+	"node": "n8n-nodes-base.mqtt",
+	"nodeVersion": "1.0",
+	"codexVersion": "1.0",
+	"categories": [
+		"Communication",
+		"Development"
+	],
+	"resources": {
+		"credentialDocumentation": [
+			{
+				"url": "https://docs.n8n.io/credentials/mqtt"
+			}
+		],
+		"primaryDocumentation": [
+			{
+				"url": "https://docs.n8n.io/nodes/n8n-nodes-base.mqtt/"
+			}
+		]
+	}
+}

packages/nodes-base/nodes/MQTT/Mqtt.node.ts

@@ -0,0 +1,171 @@
+import {
+	IExecuteFunctions,
+} from 'n8n-core';
+
+import {
+	IDataObject,
+	INodeExecutionData,
+	INodeType,
+	INodeTypeDescription,
+} from 'n8n-workflow';
+
+import * as mqtt from 'mqtt';
+
+import {
+	IClientOptions,
+} from 'mqtt';
+
+export class Mqtt implements INodeType {
+	description: INodeTypeDescription = {
+		displayName: 'MQTT',
+		name: 'mqtt',
+		icon: 'file:mqtt.svg',
+		group: ['input'],
+		version: 1,
+		description: 'Push messages to MQTT',
+		defaults: {
+			name: 'MQTT',
+			color: '#9b27af',
+		},
+		inputs: ['main'],
+		outputs: ['main'],
+		credentials: [
+			{
+				name: 'mqtt',
+				required: true,
+			},
+		],
+		properties: [
+			{
+				displayName: 'Topic',
+				name: 'topic',
+				type: 'string',
+				required: true,
+				default: '',
+				description: `The topic to publish to`,
+			},
+			{
+				displayName: 'Send Input Data',
+				name: 'sendInputData',
+				type: 'boolean',
+				default: true,
+				description: 'Send the the data the node receives as JSON.',
+			},
+			{
+				displayName: 'Message',
+				name: 'message',
+				type: 'string',
+				required: true,
+				displayOptions: {
+					show: {
+						sendInputData: [
+							false,
+						],
+					},
+				},
+				default: '',
+				description: 'The message to publish',
+			},
+			{
+				displayName: 'Options',
+				name: 'options',
+				type: 'collection',
+				placeholder: 'Add Option',
+				default: {},
+				options: [
+					{
+						displayName: 'QoS',
+						name: 'qos',
+						type: 'options',
+						options: [
+							{
+								name: 'Received at Most Once',
+								value: 0,
+							},
+							{
+								name: 'Received at Least Once',
+								value: 1,
+							},
+							{
+								name: 'Exactly Once',
+								value: 2,
+							},
+						],
+						default: 0,
+						description: 'QoS subscription level',
+					},
+					{
+						displayName: 'Retain',
+						name: 'retain',
+						type: 'boolean',
+						default: false,
+						description: `Normally if a publisher publishes a message to a topic, and no one is subscribed to<br>
+						that topic the message is simply discarded by the broker. However the publisher can tell the broker<br>
+						to keep the last message on that topic by setting the retain flag to true.`,
+					},
+				],
+			},
+		],
+	};
+
+	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+		const items = this.getInputData();
+		const length = (items.length as unknown) as number;
+		const credentials = this.getCredentials('mqtt') as IDataObject;
+
+		const protocol = credentials.protocol as string || 'mqtt';
+		const host = credentials.host as string;
+		const brokerUrl = `${protocol}://${host}`;
+		const port = credentials.port as number || 1883;
+		const clientId = credentials.clientId as string || `mqttjs_${Math.random().toString(16).substr(2, 8)}`;
+		const clean = credentials.clean as boolean;
+
+		const clientOptions: IClientOptions = {
+			port,
+			clean,
+			clientId,
+		};
+
+		if (credentials.username && credentials.password) {
+			clientOptions.username = credentials.username as string;
+			clientOptions.password = credentials.password as string;
+		}
+
+		const client = mqtt.connect(brokerUrl, clientOptions);
+		const sendInputData = this.getNodeParameter('sendInputData', 0) as boolean;
+
+		// tslint:disable-next-line: no-any
+		const data = await new Promise((resolve, reject): any => {
+			client.on('connect', () => {
+				for (let i = 0; i < length; i++) {
+
+					let message;
+					const topic = (this.getNodeParameter('topic', i) as string);
+					const options = (this.getNodeParameter('options', i) as IDataObject);
+
+					try {
+						if (sendInputData === true) {
+							message = JSON.stringify(items[i].json);
+						} else {
+							message = this.getNodeParameter('message', i) as string;
+						}
+						client.publish(topic, message, options);
+					} catch (e) {
+						reject(e);
+					}
+				}
+				//wait for the in-flight messages to be acked.
+				//needed for messages with QoS 1 & 2
+				client.end(false, {}, () => {
+					resolve([items]);
+				});
+
+				client.on('error', (e: string | undefined) => {
+					reject(e);
+				});
+			});
+		});
+
+		return data as INodeExecutionData[][];
+	}
+}

packages/nodes-base/nodes/MQTT/MqttTrigger.node.ts

@@ -13,14 +13,14 @@ import {
 import * as mqtt from 'mqtt';
 
 import {
-	IClientOptions,
+	IClientOptions, ISubscriptionMap,
 } from 'mqtt';
 
 export class MqttTrigger implements INodeType {
 	description: INodeTypeDescription = {
 		displayName: 'MQTT Trigger',
 		name: 'mqttTrigger',
-		icon: 'file:mqtt.png',
+		icon: 'file:mqtt.svg',
 		group: ['trigger'],
 		version: 1,
 		description: 'Listens to MQTT events',
@@ -43,7 +43,9 @@ export class MqttTrigger implements INodeType {
 				type: 'string',
 				default: '',
 				description: `Topics to subscribe to, multiple can be defined with comma.<br/>
-				wildcard characters are supported (+ - for single level and # - for multi level)`,
+				wildcard characters are supported (+ - for single level and # - for multi level)<br>
+				By default all subscription used QoS=0. To set a different QoS, write the QoS desired<br>
+				after the topic preceded by a colom. For Example: topicA:1,topicB:2`,
 			},
 			{
 				displayName: 'Options',
@@ -52,6 +54,13 @@ export class MqttTrigger implements INodeType {
 				placeholder: 'Add Option',
 				default: {},
 				options: [
+					{
+						displayName: 'JSON Parse Body',
+						name: 'jsonParseBody',
+						type: 'boolean',
+						default: false,
+						description: 'Try to parse the message to an object.',
+					},
 					{
 						displayName: 'Only Message',
 						name: 'onlyMessage',
@@ -59,13 +68,6 @@ export class MqttTrigger implements INodeType {
 						default: false,
 						description: 'Returns only the message property.',
 					},
-					{
-						displayName: 'JSON Parse Message',
-						name: 'jsonParseMessage',
-						type: 'boolean',
-						default: false,
-						description: 'Try to parse the message to an object.',
-					},
 				],
 			},
 		],
@@ -81,6 +83,13 @@ export class MqttTrigger implements INodeType {
 
 		const topics = (this.getNodeParameter('topics') as string).split(',');
 
+		const topicsQoS: IDataObject = {};
+
+		for (const data of topics) {
+			const [topic, qos] = data.split(':');
+			topicsQoS[topic] = (qos) ? { qos: parseInt(qos, 10) } : { qos: 0 };
+		}
+
 		const options = this.getNodeParameter('options') as IDataObject;
 
 		if (!topics) {
@@ -91,9 +100,13 @@ export class MqttTrigger implements INodeType {
 		const host = credentials.host as string;
 		const brokerUrl = `${protocol}://${host}`;
 		const port = credentials.port as number || 1883;
+		const clientId = credentials.clientId as string || `mqttjs_${Math.random().toString(16).substr(2, 8)}`;
+		const clean = credentials.clean as boolean;
 
 		const clientOptions: IClientOptions = {
 			port,
+			clean,
+			clientId,
 		};
 
 		if (credentials.username && credentials.password) {
@@ -108,20 +121,19 @@ export class MqttTrigger implements INodeType {
 		async function manualTriggerFunction() {
 			await new Promise((resolve, reject) => {
 				client.on('connect', () => {
-					client.subscribe(topics, (err, granted) => {
+					client.subscribe(topicsQoS as ISubscriptionMap, (err, granted) => {
 						if (err) {
 							reject(err);
 						}
 						client.on('message', (topic: string, message: Buffer | string) => { // tslint:disable-line:no-any
-
 							let result: IDataObject = {};
 
 							message = message.toString() as string;
 
-							if (options.jsonParseMessage) {
+							if (options.jsonParseBody) {
 								try {
 									message = JSON.parse(message.toString());
-								} catch (error) { }
+								} catch (err) { }
 							}
 
 							result.message = message;
@@ -129,10 +141,9 @@ export class MqttTrigger implements INodeType {
 
 							if (options.onlyMessage) {
 								//@ts-ignore
-								result = message;
+								result = [message as string];
 							}
-
-							self.emit([self.helpers.returnJsonArray([result])]);
+							self.emit([self.helpers.returnJsonArray(result)]);
 							resolve(true);
 						});
 					});
@@ -144,7 +155,9 @@ export class MqttTrigger implements INodeType {
 			});
 		}
 
-		manualTriggerFunction();
+		if (this.getMode() === 'trigger') {
+			manualTriggerFunction();
+		}
 
 		async function closeFunction() {
 			client.end();

packages/nodes-base/nodes/MQTT/mqtt.svg

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!-- Generator: Adobe Illustrator 24.3.0, SVG Export Plug-In . SVG Version: 6.00 Build 0)  -->
+<svg version="1.2" baseProfile="tiny" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" x="0px"
+	 y="0px" viewBox="0 0 320 320" overflow="visible" xml:space="preserve">
+<g id="black_bg" display="none">
+</g>
+<g id="logos">
+	<g>
+		<path fill="#FFFFFF" d="M7.1,133.9v46.7c73.8,0.1,134,59.3,135,132.4h45.5C186.5,214.6,106.1,134.8,7.1,133.9z"/>
+		<path fill="#FFFFFF" d="M7.1,37.3v46.7c127.4,0.1,231.1,102.5,232.1,228.9h45.5C283.7,161.4,159.7,38.3,7.1,37.3z"/>
+		<path fill="#FFFFFF" d="M312.9,193.5V97.6c-11.8-16.1-25.9-33.4-40.4-47.8c-16-15.9-34.1-30.1-52.3-42.7H119
+			C207.3,38.9,278.1,107.2,312.9,193.5z"/>
+		<path fill="#660066" d="M7.1,180.6v117.1c0,8.4,6.8,15.3,15.3,15.3H142C141,239.8,80.9,180.7,7.1,180.6z"/>
+		<path fill="#660066" d="M7.1,84.1v49.8c99,0.9,179.4,80.7,180.4,179.1h51.7C238.2,186.6,134.5,84.2,7.1,84.1z"/>
+		<path fill="#660066" d="M312.9,297.6V193.5C278.1,107.2,207.3,38.9,119,7.1H22.4c-8.4,0-15.3,6.8-15.3,15.3v15
+			c152.6,0.9,276.6,124,277.6,275.6h13C306.1,312.9,312.9,306.1,312.9,297.6z"/>
+		<path fill="#660066" d="M272.6,49.8c14.5,14.4,28.6,31.7,40.4,47.8V22.4c0-8.4-6.8-15.3-15.3-15.3h-77.3
+			C238.4,19.7,256.6,33.9,272.6,49.8z"/>
+	</g>
+</g>
+</svg>

packages/nodes-base/package.json

@@ -439,6 +439,7 @@
       "dist/nodes/Mocean/Mocean.node.js",
       "dist/nodes/MondayCom/MondayCom.node.js",
       "dist/nodes/MongoDb/MongoDb.node.js",
+      "dist/nodes/MQTT/Mqtt.node.js",
       "dist/nodes/MQTT/MqttTrigger.node.js",
       "dist/nodes/MoveBinaryData.node.js",
       "dist/nodes/Msg91/Msg91.node.js",