refactor(Redis Trigger Node): Refactor, fix duplicate triggers, and add unit tests (#9850)

packages/nodes-base/nodes/Redis/RedisTrigger.node.ts

@@ -1,6 +1,5 @@
 import type {
 	ITriggerFunctions,
-	IDataObject,
 	INodeType,
 	INodeTypeDescription,
 	ITriggerResponse,
@@ -9,6 +8,11 @@ import { NodeOperationError } from 'n8n-workflow';
 
 import { redisConnectionTest, setupRedisClient } from './utils';
 
+interface Options {
+	jsonParseBody: boolean;
+	onlyMessage: boolean;
+}
+
 export class RedisTrigger implements INodeType {
 	description: INodeTypeDescription = {
 		displayName: 'Redis Trigger',
@@ -73,45 +77,41 @@ export class RedisTrigger implements INodeType {
 		const credentials = await this.getCredentials('redis');
 
 		const channels = (this.getNodeParameter('channels') as string).split(',');
-		const options = this.getNodeParameter('options') as IDataObject;
+		const options = this.getNodeParameter('options') as Options;
 
 		if (!channels) {
 			throw new NodeOperationError(this.getNode(), 'Channels are mandatory!');
 		}
 
 		const client = setupRedisClient(credentials);
+		await client.connect();
+		await client.ping();
 
-		const manualTriggerFunction = async () => {
-			await client.connect();
-			await client.ping();
-
-			try {
-				for (const channel of channels) {
-					await client.pSubscribe(channel, (message) => {
-						if (options.jsonParseBody) {
-							try {
-								message = JSON.parse(message);
-							} catch (error) {}
-						}
-
-						if (options.onlyMessage) {
-							this.emit([this.helpers.returnJsonArray({ message })]);
-							return;
-						}
-
-						this.emit([this.helpers.returnJsonArray({ channel, message })]);
-					});
-				}
-			} catch (error) {
-				throw new NodeOperationError(this.getNode(), error);
+		const onMessage = (message: string, channel: string) => {
+			if (options.jsonParseBody) {
+				try {
+					message = JSON.parse(message);
+				} catch (error) {}
 			}
+
+			const data = options.onlyMessage ? { message } : { channel, message };
+			this.emit([this.helpers.returnJsonArray(data)]);
 		};
 
+		const manualTriggerFunction = async () =>
+			await new Promise<void>(async (resolve) => {
+				await client.pSubscribe(channels, (message, channel) => {
+					onMessage(message, channel);
+					resolve();
+				});
+			});
+
 		if (this.getMode() === 'trigger') {
-			void manualTriggerFunction();
+			await client.pSubscribe(channels, onMessage);
 		}
 
 		async function closeFunction() {
+			await client.pUnsubscribe();
 			await client.quit();
 		}