From 37d5d864e48fc4c70acd868b6611ce17e85738e9 Mon Sep 17 00:00:00 2001 From: Rupenieks Date: Wed, 5 Aug 2020 10:09:36 +0100 Subject: [PATCH 1/2] :zap: Adjusted message send format, adjusted for different formats in incoming messages --- packages/nodes-base/nodes/Amqp/Amqp.node.ts | 8 +++--- .../nodes-base/nodes/Amqp/AmqpTrigger.node.ts | 25 +++++++++++++++++-- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/packages/nodes-base/nodes/Amqp/Amqp.node.ts b/packages/nodes-base/nodes/Amqp/Amqp.node.ts index 0cde6e704f..a9bc70770c 100644 --- a/packages/nodes-base/nodes/Amqp/Amqp.node.ts +++ b/packages/nodes-base/nodes/Amqp/Amqp.node.ts @@ -57,7 +57,7 @@ export class Amqp implements INodeType { const applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object; let headerProperties = applicationProperties; - if(typeof applicationProperties === 'string' && applicationProperties !== '') { + if (typeof applicationProperties === 'string' && applicationProperties !== '') { headerProperties = JSON.parse(applicationProperties); } @@ -71,7 +71,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; @@ -83,9 +83,9 @@ export class Amqp implements INodeType { const message = { application_properties: headerProperties, - body: JSON.stringify(item) + body: item.json }; - + const sendResult = context.sender.send(message); resolve(sendResult); diff --git a/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts index 91a4eff552..d90dfb1e05 100644 --- a/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts +++ b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts @@ -5,6 +5,7 @@ import { INodeType, INodeTypeDescription, ITriggerResponse, + IDataObject, } from 'n8n-workflow'; @@ -67,12 +68,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 parseJson = this.getNodeParameter('parseJson', '') as boolean; if (sink === '') { throw new Error('Queue or Topic required!'); } + let durable = false; - if(subscription && clientname) { + + if (subscription && clientname) { durable = true; } @@ -98,7 +102,16 @@ export class AmqpTrigger implements INodeType { lastMsgId = context.message.message_id; return; } - self.emit([self.helpers.returnJsonArray([context.message])]); + + // 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])]); + } + }); const connection = container.connect(connectOptions); @@ -141,6 +154,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); }); From 47499feff71d1ac1759ccff93d83cce3ef6c0563 Mon Sep 17 00:00:00 2001 From: Jan Oberhauser Date: Thu, 6 Aug 2020 14:42:26 +0200 Subject: [PATCH 2/2] :zap: Improvements to AMQP-Node --- packages/nodes-base/nodes/Amqp/Amqp.node.ts | 30 +++++++++++-- .../nodes-base/nodes/Amqp/AmqpTrigger.node.ts | 44 ++++++++++++++----- 2 files changed, 61 insertions(+), 13 deletions(-) diff --git a/packages/nodes-base/nodes/Amqp/Amqp.node.ts b/packages/nodes-base/nodes/Amqp/Amqp.node.ts index a9bc70770c..c2803e9270 100644 --- a/packages/nodes-base/nodes/Amqp/Amqp.node.ts +++ b/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); diff --git a/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts index d90dfb1e05..9910ab5e28 100644 --- a/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts +++ b/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);