refactor(MQTT Node): Refactor, fix duplicate triggers, and add Unit tests (#9847)

Co-authored-by: Elias Meire <elias@meire.dev>

packages/nodes-base/nodes/MQTT/GenericFunctions.ts

@@ -0,0 +1,53 @@
+import { connectAsync, type IClientOptions, type MqttClient } from 'mqtt';
+import { randomString } from 'n8n-workflow';
+import { formatPrivateKey } from '@utils/utilities';
+
+interface BaseMqttCredential {
+	protocol: 'mqtt' | 'mqtts' | 'ws';
+	host: string;
+	port: number;
+	username: string;
+	password: string;
+	clean: boolean;
+	clientId: string;
+	passwordless?: boolean;
+}
+
+type NonSslMqttCredential = BaseMqttCredential & {
+	ssl: false;
+};
+
+type SslMqttCredential = BaseMqttCredential & {
+	ssl: true;
+	ca: string;
+	cert: string;
+	key: string;
+	rejectUnauthorized?: boolean;
+};
+export type MqttCredential = NonSslMqttCredential | SslMqttCredential;
+
+export const createClient = async (credentials: MqttCredential): Promise<MqttClient> => {
+	const { protocol, host, port, clean, clientId, username, password } = credentials;
+
+	const clientOptions: IClientOptions = {
+		protocol,
+		host,
+		port,
+		clean,
+		clientId: clientId || `mqttjs_${randomString(8).toLowerCase()}`,
+	};
+
+	if (username && password) {
+		clientOptions.username = username;
+		clientOptions.password = password;
+	}
+
+	if (credentials.ssl) {
+		clientOptions.ca = formatPrivateKey(credentials.ca);
+		clientOptions.cert = formatPrivateKey(credentials.cert);
+		clientOptions.key = formatPrivateKey(credentials.key);
+		clientOptions.rejectUnauthorized = credentials.rejectUnauthorized;
+	}
+
+	return await connectAsync(clientOptions);
+};

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

@@ -1,6 +1,6 @@
+import type { IClientPublishOptions } from 'mqtt';
 import type {
 	IExecuteFunctions,
-	ICredentialDataDecryptedObject,
 	ICredentialsDecrypted,
 	ICredentialTestFunctions,
 	INodeCredentialTestResult,
@@ -8,10 +8,10 @@ import type {
 	INodeType,
 	INodeTypeDescription,
 } from 'n8n-workflow';
-import { randomString } from 'n8n-workflow';
 
-import * as mqtt from 'mqtt';
-import { formatPrivateKey } from '@utils/utilities';
+import { createClient, type MqttCredential } from './GenericFunctions';
+
+type PublishOption = Pick<IClientPublishOptions, 'qos' | 'retain'>;
 
 export class Mqtt implements INodeType {
 	description: INodeTypeDescription = {
@@ -110,63 +110,11 @@ export class Mqtt implements INodeType {
 				this: ICredentialTestFunctions,
 				credential: ICredentialsDecrypted,
 			): Promise<INodeCredentialTestResult> {
-				const credentials = credential.data as ICredentialDataDecryptedObject;
+				const credentials = credential.data as unknown as MqttCredential;
+
 				try {
-					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_${randomString(8).toLowerCase()}`;
-					const clean = credentials.clean as boolean;
-					const ssl = credentials.ssl as boolean;
-					const ca = formatPrivateKey(credentials.ca as string);
-					const cert = formatPrivateKey(credentials.cert as string);
-					const key = formatPrivateKey(credentials.key as string);
-					const rejectUnauthorized = credentials.rejectUnauthorized as boolean;
-
-					let client: mqtt.MqttClient;
-
-					if (!ssl) {
-						const clientOptions: mqtt.IClientOptions = {
-							port,
-							clean,
-							clientId,
-						};
-
-						if (credentials.username && credentials.password) {
-							clientOptions.username = credentials.username as string;
-							clientOptions.password = credentials.password as string;
-						}
-						client = mqtt.connect(brokerUrl, clientOptions);
-					} else {
-						const clientOptions: mqtt.IClientOptions = {
-							port,
-							clean,
-							clientId,
-							ca,
-							cert,
-							key,
-							rejectUnauthorized,
-						};
-						if (credentials.username && credentials.password) {
-							clientOptions.username = credentials.username as string;
-							clientOptions.password = credentials.password as string;
-						}
-
-						client = mqtt.connect(brokerUrl, clientOptions);
-					}
-
-					await new Promise((resolve, reject) => {
-						client.on('connect', (test) => {
-							resolve(test);
-							client.end();
-						});
-						client.on('error', (error) => {
-							client.end();
-							reject(error);
-						});
-					});
+					const client = await createClient(credentials);
+					client.end();
 				} catch (error) {
 					return {
 						status: 'Error',
@@ -182,87 +130,27 @@ export class Mqtt implements INodeType {
 	};
 
 	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+		const credentials = (await this.getCredentials('mqtt')) as unknown as MqttCredential;
+		const client = await createClient(credentials);
+
+		const publishPromises = [];
 		const items = this.getInputData();
-		const length = items.length;
-		const credentials = await this.getCredentials('mqtt');
-
-		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_${randomString(8).toLowerCase()}`;
-		const clean = credentials.clean as boolean;
-		const ssl = credentials.ssl as boolean;
-		const ca = credentials.ca as string;
-		const cert = credentials.cert as string;
-		const key = credentials.key as string;
-		const rejectUnauthorized = credentials.rejectUnauthorized as boolean;
-
-		let client: mqtt.MqttClient;
-
-		if (!ssl) {
-			const clientOptions: mqtt.IClientOptions = {
-				port,
-				clean,
-				clientId,
-			};
-
-			if (credentials.username && credentials.password) {
-				clientOptions.username = credentials.username as string;
-				clientOptions.password = credentials.password as string;
-			}
-
-			client = mqtt.connect(brokerUrl, clientOptions);
-		} else {
-			const clientOptions: mqtt.IClientOptions = {
-				port,
-				clean,
-				clientId,
-				ca,
-				cert,
-				key,
-				rejectUnauthorized,
-			};
-			if (credentials.username && credentials.password) {
-				clientOptions.username = credentials.username as string;
-				clientOptions.password = credentials.password as string;
-			}
-
-			client = mqtt.connect(brokerUrl, clientOptions);
+		for (let i = 0; i < items.length; i++) {
+			const topic = this.getNodeParameter('topic', i) as string;
+			const options = this.getNodeParameter('options', i) as unknown as PublishOption;
+			const sendInputData = this.getNodeParameter('sendInputData', i) as boolean;
+			const message = sendInputData
+				? JSON.stringify(items[i].json)
+				: (this.getNodeParameter('message', i) as string);
+			publishPromises.push(client.publishAsync(topic, message, options));
 		}
 
-		const sendInputData = this.getNodeParameter('sendInputData', 0) as boolean;
+		await Promise.all(publishPromises);
 
-		const data = await new Promise((resolve, reject) => {
-			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);
+		// wait for the in-flight messages to be acked.
+		// needed for messages with QoS 1 & 2
+		await client.endAsync();
 
-					try {
-						if (sendInputData) {
-							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) => {
-					reject(e);
-				});
-			});
-		});
-
-		return data as INodeExecutionData[][];
+		return [items];
 	}
 }

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

@@ -1,16 +1,22 @@
+import type { ISubscriptionMap } from 'mqtt';
+import type { QoS } from 'mqtt-packet';
 import type {
 	ITriggerFunctions,
 	IDataObject,
 	INodeType,
 	INodeTypeDescription,
 	ITriggerResponse,
-	IDeferredPromise,
 	IRun,
 } from 'n8n-workflow';
-import { NodeOperationError, randomString } from 'n8n-workflow';
+import { NodeOperationError } from 'n8n-workflow';
 
-import * as mqtt from 'mqtt';
-import { formatPrivateKey } from '@utils/utilities';
+import { createClient, type MqttCredential } from './GenericFunctions';
+
+interface Options {
+	jsonParseBody: boolean;
+	onlyMessage: boolean;
+	parallelProcessing: boolean;
+}
 
 export class MqttTrigger implements INodeType {
 	description: INodeTypeDescription = {
@@ -87,120 +93,64 @@ export class MqttTrigger implements INodeType {
 	};
 
 	async trigger(this: ITriggerFunctions): Promise<ITriggerResponse> {
-		const credentials = await this.getCredentials('mqtt');
-
 		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;
-		const parallelProcessing = this.getNodeParameter('options.parallelProcessing', true) as boolean;
-
-		if (!topics) {
+		if (!topics?.length) {
 			throw new NodeOperationError(this.getNode(), 'Topics are mandatory!');
 		}
 
-		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_${randomString(8).toLowerCase()}`;
-		const clean = credentials.clean as boolean;
-		const ssl = credentials.ssl as boolean;
-		const ca = formatPrivateKey(credentials.ca as string);
-		const cert = formatPrivateKey(credentials.cert as string);
-		const key = formatPrivateKey(credentials.key as string);
-		const rejectUnauthorized = credentials.rejectUnauthorized as boolean;
-
-		let client: mqtt.MqttClient;
-
-		if (!ssl) {
-			const clientOptions: mqtt.IClientOptions = {
-				port,
-				clean,
-				clientId,
-			};
-
-			if (credentials.username && credentials.password) {
-				clientOptions.username = credentials.username as string;
-				clientOptions.password = credentials.password as string;
-			}
-
-			client = mqtt.connect(brokerUrl, clientOptions);
-		} else {
-			const clientOptions: mqtt.IClientOptions = {
-				port,
-				clean,
-				clientId,
-				ca,
-				cert,
-				key,
-				rejectUnauthorized,
-			};
-			if (credentials.username && credentials.password) {
-				clientOptions.username = credentials.username as string;
-				clientOptions.password = credentials.password as string;
-			}
-
-			client = mqtt.connect(brokerUrl, clientOptions);
+		const topicsQoS: ISubscriptionMap = {};
+		for (const data of topics) {
+			const [topic, qosString] = data.split(':');
+			let qos = qosString ? parseInt(qosString, 10) : 0;
+			if (qos < 0 || qos > 2) qos = 0;
+			topicsQoS[topic] = { qos: qos as QoS };
 		}
 
-		const manualTriggerFunction = async () => {
-			await new Promise((resolve, reject) => {
-				client.on('connect', () => {
-					client.subscribe(topicsQoS as mqtt.ISubscriptionMap, (error, _granted) => {
-						if (error) {
-							reject(error);
-						}
-						client.on('message', async (topic: string, message: Buffer | string) => {
-							let result: IDataObject = {};
+		const options = this.getNodeParameter('options') as Options;
+		const credentials = (await this.getCredentials('mqtt')) as unknown as MqttCredential;
+		const client = await createClient(credentials);
 
-							message = message.toString();
+		const parsePayload = (topic: string, payload: Buffer) => {
+			let message = payload.toString();
 
-							if (options.jsonParseBody) {
-								try {
-									message = JSON.parse(message.toString());
-								} catch (e) {}
-							}
+			if (options.jsonParseBody) {
+				try {
+					message = JSON.parse(message);
+				} catch (e) {}
+			}
 
-							result.message = message;
-							result.topic = topic;
+			let result: IDataObject = { message, topic };
 
-							if (options.onlyMessage) {
-								//@ts-ignore
-								result = [message as string];
-							}
+			if (options.onlyMessage) {
+				//@ts-ignore
+				result = [message];
+			}
 
-							let responsePromise: IDeferredPromise<IRun> | undefined;
-							if (!parallelProcessing) {
-								responsePromise = await this.helpers.createDeferredPromise();
-							}
-							this.emit([this.helpers.returnJsonArray([result])], undefined, responsePromise);
-							if (responsePromise) {
-								await responsePromise.promise();
-							}
-							resolve(true);
-						});
-					});
-				});
-
-				client.on('error', (error) => {
-					reject(error);
-				});
-			});
+			return [this.helpers.returnJsonArray([result])];
 		};
 
+		const manualTriggerFunction = async () =>
+			await new Promise<void>(async (resolve) => {
+				client.once('message', (topic, payload) => {
+					this.emit(parsePayload(topic, payload));
+					resolve();
+				});
+				await client.subscribeAsync(topicsQoS);
+			});
+
 		if (this.getMode() === 'trigger') {
-			void manualTriggerFunction();
+			const donePromise = !options.parallelProcessing
+				? await this.helpers.createDeferredPromise<IRun>()
+				: undefined;
+			client.on('message', async (topic, payload) => {
+				this.emit(parsePayload(topic, payload), undefined, donePromise);
+				await donePromise?.promise();
+			});
+			await client.subscribeAsync(topicsQoS);
 		}
 
 		async function closeFunction() {
-			client.end();
+			await client.endAsync();
 		}
 
 		return {

packages/nodes-base/nodes/MQTT/test/GenericFunctions.test.ts

@@ -0,0 +1,38 @@
+import { MqttClient } from 'mqtt';
+import { mock } from 'jest-mock-extended';
+
+import { createClient, type MqttCredential } from '../GenericFunctions';
+
+describe('createClient', () => {
+	const mockConnect = jest.spyOn(MqttClient.prototype, 'connect').mockImplementation(function (
+		this: MqttClient,
+	) {
+		setImmediate(() => this.emit('connect', mock()));
+		return this;
+	});
+
+	beforeEach(() => jest.clearAllMocks());
+
+	it('should create a client with minimal credentials', async () => {
+		const credentials = mock<MqttCredential>({
+			protocol: 'mqtt',
+			host: 'localhost',
+			port: 1883,
+			clean: true,
+			clientId: 'testClient',
+			ssl: false,
+		});
+		const client = await createClient(credentials);
+
+		expect(mockConnect).toBeCalledTimes(1);
+		expect(client).toBeDefined();
+		expect(client).toBeInstanceOf(MqttClient);
+		expect(client.options).toMatchObject({
+			protocol: 'mqtt',
+			host: 'localhost',
+			port: 1883,
+			clean: true,
+			clientId: 'testClient',
+		});
+	});
+});

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

@@ -0,0 +1,56 @@
+import type { MqttClient } from 'mqtt';
+import { mock } from 'jest-mock-extended';
+import type { ICredentialDataDecryptedObject, IExecuteFunctions } from 'n8n-workflow';
+
+import { Mqtt } from '../Mqtt.node';
+import { createClient } from '../GenericFunctions';
+
+jest.mock('../GenericFunctions', () => {
+	const mockMqttClient = mock<MqttClient>();
+	return {
+		createClient: jest.fn().mockResolvedValue(mockMqttClient),
+	};
+});
+
+describe('MQTT Node', () => {
+	const credentials = mock<ICredentialDataDecryptedObject>();
+	const executeFunctions = mock<IExecuteFunctions>();
+
+	beforeEach(() => {
+		jest.clearAllMocks();
+
+		executeFunctions.getCredentials.calledWith('mqtt').mockResolvedValue(credentials);
+		executeFunctions.getInputData.mockReturnValue([{ json: { testing: true } }]);
+		executeFunctions.getNodeParameter.calledWith('topic', 0).mockReturnValue('test/topic');
+		executeFunctions.getNodeParameter.calledWith('options', 0).mockReturnValue({});
+	});
+
+	it('should publish input data', async () => {
+		executeFunctions.getNodeParameter.calledWith('sendInputData', 0).mockReturnValue(true);
+
+		const result = await new Mqtt().execute.call(executeFunctions);
+
+		expect(result).toEqual([[{ json: { testing: true } }]]);
+		expect(executeFunctions.getCredentials).toHaveBeenCalledTimes(1);
+		expect(executeFunctions.getNodeParameter).toHaveBeenCalledTimes(3);
+
+		const mockMqttClient = await createClient(mock());
+		expect(mockMqttClient.publishAsync).toHaveBeenCalledWith('test/topic', '{"testing":true}', {});
+		expect(mockMqttClient.endAsync).toHaveBeenCalledTimes(1);
+	});
+
+	it('should publish a custom message', async () => {
+		executeFunctions.getNodeParameter.calledWith('sendInputData', 0).mockReturnValue(false);
+		executeFunctions.getNodeParameter.calledWith('message', 0).mockReturnValue('Hello, MQTT!');
+
+		const result = await new Mqtt().execute.call(executeFunctions);
+
+		expect(result).toEqual([[{ json: { testing: true } }]]);
+		expect(executeFunctions.getCredentials).toHaveBeenCalledTimes(1);
+		expect(executeFunctions.getNodeParameter).toHaveBeenCalledTimes(4);
+
+		const mockMqttClient = await createClient(mock());
+		expect(mockMqttClient.publishAsync).toHaveBeenCalledWith('test/topic', 'Hello, MQTT!', {});
+		expect(mockMqttClient.endAsync).toHaveBeenCalledTimes(1);
+	});
+});

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

@@ -0,0 +1,123 @@
+import type { MqttClient, OnMessageCallback } from 'mqtt';
+import { returnJsonArray } from 'n8n-core';
+import { captor, mock } from 'jest-mock-extended';
+import type { ICredentialDataDecryptedObject, ITriggerFunctions } from 'n8n-workflow';
+
+import { MqttTrigger } from '../MqttTrigger.node';
+import { createClient } from '../GenericFunctions';
+
+jest.mock('../GenericFunctions', () => {
+	const mockMqttClient = mock<MqttClient>();
+	return {
+		createClient: jest.fn().mockResolvedValue(mockMqttClient),
+	};
+});
+
+describe('MQTT Trigger Node', () => {
+	const topic = 'test/topic';
+	const payload = Buffer.from('{"testing": true}');
+	const credentials = mock<ICredentialDataDecryptedObject>();
+	const triggerFunctions = mock<ITriggerFunctions>({
+		helpers: { returnJsonArray },
+	});
+
+	beforeEach(() => {
+		jest.clearAllMocks();
+
+		triggerFunctions.getCredentials.calledWith('mqtt').mockResolvedValue(credentials);
+		triggerFunctions.getNodeParameter.calledWith('topics').mockReturnValue(topic);
+	});
+
+	it('should emit in manual mode', async () => {
+		triggerFunctions.getMode.mockReturnValue('manual');
+		triggerFunctions.getNodeParameter.calledWith('options').mockReturnValue({});
+
+		const response = await new MqttTrigger().trigger.call(triggerFunctions);
+		expect(response.manualTriggerFunction).toBeDefined();
+		expect(response.closeFunction).toBeDefined();
+
+		expect(triggerFunctions.getCredentials).toHaveBeenCalledTimes(1);
+		expect(triggerFunctions.getNodeParameter).toHaveBeenCalledTimes(2);
+
+		// manually trigger the node, like Workflow.runNode does
+		const triggerPromise = response.manualTriggerFunction!();
+
+		const mockMqttClient = await createClient(mock());
+		expect(mockMqttClient.on).not.toHaveBeenCalled();
+
+		const onMessageCaptor = captor<OnMessageCallback>();
+		expect(mockMqttClient.once).toHaveBeenCalledWith('message', onMessageCaptor);
+		expect(mockMqttClient.subscribeAsync).toHaveBeenCalledWith({ [topic]: { qos: 0 } });
+		expect(triggerFunctions.emit).not.toHaveBeenCalled();
+
+		// simulate a message
+		const onMessage = onMessageCaptor.value;
+		onMessage('test/topic', payload, mock());
+		expect(triggerFunctions.emit).toHaveBeenCalledWith([
+			[{ json: { message: '{"testing": true}', topic } }],
+		]);
+
+		// wait for the promise to resolve
+		await new Promise((resolve) => setImmediate(resolve));
+		await expect(triggerPromise).resolves.toEqual(undefined);
+
+		expect(mockMqttClient.endAsync).not.toHaveBeenCalled();
+		await response.closeFunction!();
+		expect(mockMqttClient.endAsync).toHaveBeenCalledTimes(1);
+	});
+
+	it('should emit in trigger mode', async () => {
+		triggerFunctions.getMode.mockReturnValue('trigger');
+		triggerFunctions.getNodeParameter.calledWith('options').mockReturnValue({});
+
+		const response = await new MqttTrigger().trigger.call(triggerFunctions);
+		expect(response.manualTriggerFunction).toBeDefined();
+		expect(response.closeFunction).toBeDefined();
+
+		expect(triggerFunctions.getCredentials).toHaveBeenCalledTimes(1);
+		expect(triggerFunctions.getNodeParameter).toHaveBeenCalledTimes(2);
+
+		const mockMqttClient = await createClient(mock());
+		expect(mockMqttClient.once).not.toHaveBeenCalled();
+
+		const onMessageCaptor = captor<OnMessageCallback>();
+		expect(mockMqttClient.on).toHaveBeenCalledWith('message', onMessageCaptor);
+		expect(mockMqttClient.subscribeAsync).toHaveBeenCalledWith({ [topic]: { qos: 0 } });
+		expect(triggerFunctions.emit).not.toHaveBeenCalled();
+
+		// simulate a message
+		const onMessage = onMessageCaptor.value;
+		onMessage('test/topic', payload, mock());
+		expect(triggerFunctions.emit).toHaveBeenCalledWith(
+			[[{ json: { message: '{"testing": true}', topic } }]],
+			undefined,
+			undefined,
+		);
+
+		expect(mockMqttClient.endAsync).not.toHaveBeenCalled();
+		await response.closeFunction!();
+		expect(mockMqttClient.endAsync).toHaveBeenCalledTimes(1);
+	});
+
+	it('should parse JSON messages when configured', async () => {
+		triggerFunctions.getMode.mockReturnValue('trigger');
+		triggerFunctions.getNodeParameter.calledWith('options').mockReturnValue({
+			jsonParseBody: true,
+		});
+
+		await new MqttTrigger().trigger.call(triggerFunctions);
+
+		const mockMqttClient = await createClient(mock());
+		const onMessageCaptor = captor<OnMessageCallback>();
+		expect(mockMqttClient.on).toHaveBeenCalledWith('message', onMessageCaptor);
+
+		// simulate a message
+		const onMessage = onMessageCaptor.value;
+		onMessage('test/topic', payload, mock());
+		expect(triggerFunctions.emit).toHaveBeenCalledWith(
+			[[{ json: { message: { testing: true }, topic } }]],
+			undefined,
+			undefined,
+		);
+	});
+});

packages/nodes-base/package.json

@@ -873,7 +873,7 @@
     "minifaker": "1.34.1",
     "moment-timezone": "0.5.37",
     "mongodb": "6.3.0",
-    "mqtt": "5.0.2",
+    "mqtt": "5.7.2",
     "mssql": "10.0.2",
     "mysql2": "3.10.0",
     "n8n-workflow": "workspace:*",