🔀 Merge branch 'Fix-AMQP-Nodes'

packages/nodes-base/nodes/Amqp/Amqp.node.ts

@@ -2,6 +2,7 @@ import { ContainerOptions, Delivery } from 'rhea';
 
 import { IExecuteSingleFunctions } from 'n8n-core';
 import {
+	IDataObject,
 	INodeExecutionData,
 	INodeType,
 	INodeTypeDescription,
@@ -41,7 +42,23 @@ export class Amqp implements INodeType {
 				type: 'json',
 				default: '',
 				description: 'Header parameters as JSON (flat object). Sent as application_properties in amqp-message meta info.',
-			}
+			},
+			{
+				displayName: 'Options',
+				name: 'options',
+				type: 'collection',
+				placeholder: 'Add Option',
+				default: {},
+				options: [
+					{
+						displayName: 'Data as Object',
+						name: 'dataAsObject',
+						type: 'boolean',
+						default: false,
+						description: 'Send the data as an object.',
+					},
+				],
+			},
 		]
 	};
 
@@ -55,9 +72,10 @@ export class Amqp implements INodeType {
 
 		const sink = this.getNodeParameter('sink', '') as string;
 		const applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object;
+		const options = this.getNodeParameter('options', {}) as IDataObject;
 
 		let headerProperties = applicationProperties;
-		if(typeof applicationProperties === 'string' && applicationProperties !== '') {
+		if (typeof applicationProperties === 'string' && applicationProperties !== '') {
 			headerProperties = JSON.parse(applicationProperties);
 		}
 
@@ -71,7 +89,7 @@ export class Amqp implements INodeType {
 			host: credentials.hostname,
 			port: credentials.port,
 			reconnect: true,		// this id the default anyway
-			reconnect_limit: 50,	// try for max 50 times, based on a back-off algorithm
+			reconnect_limit: 50, 	// try for max 50 times, based on a back-off algorithm
 		};
 		if (credentials.username || credentials.password) {
 			container.options.username = credentials.username;
@@ -81,9 +99,15 @@ export class Amqp implements INodeType {
 		const allSent = new Promise(( resolve ) => {
 			container.on('sendable', (context: any) => { // tslint:disable-line:no-any
 
+				let body: IDataObject | string = item.json;
+
+				if (options.dataAsObject !== true) {
+					body = JSON.stringify(body);
+				}
+
 				const message = {
 					application_properties: headerProperties,
-					body: JSON.stringify(item)
+					body
 				};
 
 				const sendResult = context.sender.send(message);

packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts

@@ -2,6 +2,7 @@ import { ContainerOptions } from 'rhea';
 
 import { ITriggerFunctions } from 'n8n-core';
 import {
+	IDataObject,
 	INodeType,
 	INodeTypeDescription,
 	ITriggerResponse,
@@ -53,6 +54,29 @@ export class AmqpTrigger implements INodeType {
 				placeholder: 'for durable/persistent topic subscriptions, example: "order-worker"',
 				description: 'Leave empty for non-durable topic subscriptions or queues',
 			},
+			{
+				displayName: 'Options',
+				name: 'options',
+				type: 'collection',
+				placeholder: 'Add Option',
+				default: {},
+				options: [
+					{
+						displayName: 'Only Body',
+						name: 'onlyBody',
+						type: 'boolean',
+						default: false,
+						description: 'Returns only the body property.',
+					},
+					{
+						displayName: 'JSON Parse Body',
+						name: 'jsonParseBody',
+						type: 'boolean',
+						default: false,
+						description: 'Parse the body to an object.',
+					},
+				],
+			},
 		]
 	};
 
@@ -67,12 +91,15 @@ export class AmqpTrigger implements INodeType {
 		const sink = this.getNodeParameter('sink', '') as string;
 		const clientname = this.getNodeParameter('clientname', '') as string;
 		const subscription = this.getNodeParameter('subscription', '') as string;
+		const options = this.getNodeParameter('options', {}) as IDataObject;
 
 		if (sink === '') {
 			throw new Error('Queue or Topic required!');
 		}
+
 		let durable = false;
-		if(subscription && clientname) {
+
+		if (subscription && clientname) {
 			durable = true;
 		}
 
@@ -98,7 +125,17 @@ export class AmqpTrigger implements INodeType {
 				lastMsgId = context.message.message_id;
 				return;
 			}
-			self.emit([self.helpers.returnJsonArray([context.message])]);
+
+			let data = context.message;
+
+			if (options.jsonParseBody === true) {
+				data.body = JSON.parse(data.body);
+			}
+			if (options.onlyBody === true) {
+				data = data.body;
+			}
+
+			self.emit([self.helpers.returnJsonArray([data])]);
 		});
 
 		const connection = container.connect(connectOptions);
@@ -141,6 +178,14 @@ export class AmqpTrigger implements INodeType {
 					reject(new Error('Aborted, no message received within 30secs. This 30sec timeout is only set for "manually triggered execution". Active Workflows will listen indefinitely.'));
 				}, 30000);
 				container.on('message', (context: any) => { // tslint:disable-line:no-any
+					// Check if the only property present in the message is body
+					// in which case we only emit the content of the body property
+					// otherwise we emit all properties and their content
+					if (Object.keys(context.message)[0] === 'body' && Object.keys(context.message).length === 1) {
+						self.emit([self.helpers.returnJsonArray([context.message.body])]);
+					} else {
+						self.emit([self.helpers.returnJsonArray([context.message])]);
+					}
 					clearTimeout(timeoutHandler);
 					resolve(true);
 				});