⚡ Improvements to AMQP-Node

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,6 +72,7 @@ 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 !== '') {
@@ -81,11 +99,17 @@ 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: item.json
+					body
 				};
-				
+
 				const sendResult = context.sender.send(message);
 
 				resolve(sendResult);

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

@@ -2,10 +2,10 @@ import { ContainerOptions } from 'rhea';
 
 import { ITriggerFunctions } from 'n8n-core';
 import {
+	IDataObject,
 	INodeType,
 	INodeTypeDescription,
 	ITriggerResponse,
-	IDataObject,
 } from 'n8n-workflow';
 
 
@@ -54,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.',
+					},
+				],
+			},
 		]
 	};
 
@@ -68,7 +91,7 @@ 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 parseJson = this.getNodeParameter('parseJson', '') as boolean;
+		const options = this.getNodeParameter('options', {}) as IDataObject;
 
 		if (sink === '') {
 			throw new Error('Queue or Topic required!');
@@ -103,15 +126,16 @@ export class AmqpTrigger implements INodeType {
 				return;
 			}
 
-			// 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])]);
+			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);