⚡ Add acknowledge option for RabbitMQ

packages/nodes-base/nodes/RabbitMQ/RabbitMQRespondToTrigger.node.json

@@ -0,0 +1,19 @@
+{
+	"node": "n8n-nodes-base.rabbitmqRespondToTrigger",
+	"nodeVersion": "1.0",
+	"codexVersion": "1.0",
+	"categories": [
+		"Core Nodes",
+		"Utility"
+	],
+	"resources": {
+		"primaryDocumentation": [
+			{
+				"url": "https://docs.n8n.io/nodes/n8n-nodes-base.rabbitmqRespondToTrigger/"
+			}
+		]
+	},
+	"subcategories": {
+			"Core Nodes":["Flow"]
+		}
+}

packages/nodes-base/nodes/RabbitMQ/RabbitMQRespondToTrigger.node.ts

@@ -0,0 +1,50 @@
+import {
+	IExecuteFunctions,
+} from 'n8n-core';
+
+import {
+	IDataObject,
+	IN8nHttpFullResponse,
+	IN8nHttpResponse,
+	INodeExecutionData,
+	INodeType,
+	INodeTypeDescription,
+} from 'n8n-workflow';
+
+export class RabbitMQRespondToTrigger implements INodeType {
+	description: INodeTypeDescription = {
+		displayName: 'RabbitMQ Respond to Trigger',
+		icon: 'file:rabbitmq.png',
+		name: 'rabbitmqRespondToTrigger',
+		group: ['transform'],
+		version: 1,
+		description: 'Returns data for Trigger',
+		defaults: {
+			name: 'RabbitMQ Respond to Trigger',
+		},
+		inputs: ['main'],
+		outputs: ['main'],
+		credentials: [
+		],
+		properties: [
+			{
+				displayName: 'Acknowledge',
+				name: 'acknowledge',
+				type: 'boolean',
+				default: true,
+				description: 'Set \'true\' for acknowledge or \'false\' for not acknowledge.',
+			},
+		],
+	};
+
+	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+		const items = this.getInputData();
+
+		const acknowledge = this.getNodeParameter('acknowledge', 0, false) as boolean;
+		const response = {ack: acknowledge} as IDataObject;
+
+		this.sendResponse(response);
+
+		return this.prepareOutputData(items);
+	}
+}

packages/nodes-base/nodes/RabbitMQ/RabbitMQTrigger.node.ts

@@ -6,6 +6,9 @@ import {
 	INodeTypeDescription,
 	ITriggerFunctions,
 	ITriggerResponse,
+	IDeferredPromise,
+	IExecuteResponsePromiseData,
+	createDeferredPromise,
 } from 'n8n-workflow';
 
 import {
@@ -44,7 +47,38 @@ export class RabbitMQTrigger implements INodeType {
 				placeholder: 'queue-name',
 				description: 'Name of the queue to publish to',
 			},
-
+			{
+				displayName: 'Acknowledge',
+				name: 'acknowledgeMode',
+				type: 'options',
+				options: [
+					{
+						name: 'Immediately',
+						value: 'onReceived',
+						description: 'As soon as this node executes',
+					},
+					{
+						name: 'Using \'RabbitMQ Respond to Trigger\' node',
+						value: 'acknowledgeNode',
+						description: 'Acknowledge when acknowledge node executes',
+					},
+				],
+				default: 'onReceived',
+				description: 'When and how to acknowledge to the trigger.',
+			},
+			{
+				displayName: 'Insert a \'RabbitMQ Respond to Trigger\' node to control when and how you acknowledge. <a href="https://docs.n8n.io/nodes/n8n-nodes-base.rabbitmqRespondToTrigger" target="_blank">More details</a>',
+				name: 'triggerNotice',
+				type: 'notice',
+				displayOptions: {
+					show: {
+						acknowledgeMode: [
+							'acknowledgeNode',
+						],
+					},
+				},
+				default: '',
+			},
 			{
 				displayName: 'Options',
 				name: 'options',
@@ -101,6 +135,7 @@ export class RabbitMQTrigger implements INodeType {
 	async trigger(this: ITriggerFunctions): Promise<ITriggerResponse> {
 		const queue = this.getNodeParameter('queue') as string;
 		const options = this.getNodeParameter('options', {}) as IDataObject;
+		const acknowledgeMode = this.getNodeParameter('acknowledgeMode') as string;
 
 		const channel = await rabbitmqConnectQueue.call(this, queue, options);
 
@@ -134,12 +169,36 @@ export class RabbitMQTrigger implements INodeType {
 						}
 					}
 
-					self.emit([
-						[
-							item,
-						],
-					]);
-					channel.ack(message);
+					if (acknowledgeMode == 'onReceived') {
+						self.emit([
+							[
+								item,
+							],
+						]);
+
+						channel.ack(message);
+					} else {
+						let acknowledgePromise = await createDeferredPromise<IExecuteResponsePromiseData>();
+						acknowledgePromise
+							.promise()
+							.then((response: IExecuteResponsePromiseData) => {
+								let acknowledgeResponse = response as IDataObject;
+								if (acknowledgeResponse.ack) {
+									channel.ack(message);
+								} else {
+									channel.nack(message);
+								}
+							})
+							.catch(async () => {
+								channel.nack(message);
+							});
+
+						self.emit([
+							[
+								item,
+							],
+						], acknowledgePromise);
+					}
 				}
 			});
 		};
@@ -168,5 +227,4 @@ export class RabbitMQTrigger implements INodeType {
 			manualTriggerFunction,
 		};
 	}
-
 }

packages/nodes-base/package.json

@@ -583,6 +583,7 @@
       "dist/nodes/QuickBase/QuickBase.node.js",
       "dist/nodes/QuickBooks/QuickBooks.node.js",
       "dist/nodes/RabbitMQ/RabbitMQ.node.js",
+      "dist/nodes/RabbitMQ/RabbitMQRespondToTrigger.node.js",
       "dist/nodes/RabbitMQ/RabbitMQTrigger.node.js",
       "dist/nodes/Raindrop/Raindrop.node.js",
       "dist/nodes/ReadBinaryFile/ReadBinaryFile.node.js",