From c31dc77b06f8b84d3b8323b2affb278851276736 Mon Sep 17 00:00:00 2001 From: Leo Date: Fri, 1 Nov 2019 13:50:33 +0100 Subject: [PATCH] added amqp sender --- .../credentials/Amqp.credentials.ts | 12 ++ packages/nodes-base/nodes/Amqp/Amqp.node.ts | 117 ++++++++++++++++++ ...qpListener.node.ts => AmqpTrigger.node.ts} | 64 ++++------ packages/nodes-base/package.json | 3 +- 4 files changed, 157 insertions(+), 39 deletions(-) create mode 100644 packages/nodes-base/nodes/Amqp/Amqp.node.ts rename packages/nodes-base/nodes/Amqp/{AmqpListener.node.ts => AmqpTrigger.node.ts} (70%) diff --git a/packages/nodes-base/credentials/Amqp.credentials.ts b/packages/nodes-base/credentials/Amqp.credentials.ts index ec0c57014c..640dedb18b 100644 --- a/packages/nodes-base/credentials/Amqp.credentials.ts +++ b/packages/nodes-base/credentials/Amqp.credentials.ts @@ -11,6 +11,18 @@ export class Amqp implements ICredentialType { // The credentials to get from user and save encrypted. // Properties can be defined exactly in the same way // as node properties. + { + displayName: 'Hostname', + name: 'hostname', + type: 'string' as NodePropertyTypes, + default: '', + }, + { + displayName: 'Port', + name: 'port', + type: 'number' as NodePropertyTypes, + default: 5672, + }, { displayName: 'User', name: 'username', diff --git a/packages/nodes-base/nodes/Amqp/Amqp.node.ts b/packages/nodes-base/nodes/Amqp/Amqp.node.ts new file mode 100644 index 0000000000..9dcb3ca5ef --- /dev/null +++ b/packages/nodes-base/nodes/Amqp/Amqp.node.ts @@ -0,0 +1,117 @@ +import { IExecuteSingleFunctions } from 'n8n-core'; +import { + IDataObject, + INodeExecutionData, + INodeType, + INodeTypeDescription, +} from 'n8n-workflow'; +import { Delivery } from 'rhea'; + +export class Amqp implements INodeType { + description: INodeTypeDescription = { + displayName: 'AMQP Sender', + name: 'amqpSender', + icon: 'file:amqp.png', + group: ['transform'], + version: 1, + description: 'Sends a raw-message via AMQP 1.0, executed once per item', + defaults: { + name: 'AMQP Sender', + color: '#00FF00', + }, + inputs: ['main'], + outputs: ['main'], + credentials: [{ + name: 'amqp', + required: true, + }], + properties: [ + { + displayName: 'Host', + name: 'hostname', + type: 'string', + default: 'localhost', + description: 'hostname of the amqp server', + }, + { + displayName: 'Port', + name: 'port', + type: 'number', + typeOptions: { + minValue: 1, + }, + default: 5672, + description: 'TCP Port to connect to', + }, + { + displayName: 'Queue / Topic', + name: 'sink', + type: 'string', + default: '', + placeholder: 'topic://sourcename.something', + description: 'name of the queue of topic to publish to', + }, + // Header Parameters + { + displayName: 'Headers', + name: 'headerParametersJson', + type: 'json', + default: '', + description: 'Header parameters as JSON (flat object). Sent as application_properties in amqp-message meta info.', + } + ] + }; + + async executeSingle(this: IExecuteSingleFunctions): Promise { + const item = this.getInputData(); + + const credentials = this.getCredentials('amqp'); + if (!credentials) { + throw new Error('Credentials are mandatory!'); + } + + const sink = this.getNodeParameter('sink', '') as string; + let applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object; + + let headerProperties = applicationProperties; + if(typeof applicationProperties === 'string' && applicationProperties != '') { + headerProperties = JSON.parse(applicationProperties) + } + + if (sink == '') { + throw new Error('Queue or Topic required!'); + } + + let container = require('rhea'); + + let connectOptions = { + 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 + } + if (credentials.username || credentials.password) { + container.options.username = credentials.username; + container.options.password = credentials.password; + } + + let allSent = new Promise( function( resolve ) { + container.on('sendable', function (context: any) { + + let message = { + application_properties: headerProperties, + body: JSON.stringify(item) + } + let sendResult = context.sender.send(message); + + resolve(sendResult); + }); + }); + + container.connect(connectOptions).open_sender(sink); + + let sendResult: Delivery = await allSent as Delivery; // sendResult has a a property that causes circular reference if returned + + return { json: { id: sendResult.id } } as INodeExecutionData; + } +} diff --git a/packages/nodes-base/nodes/Amqp/AmqpListener.node.ts b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts similarity index 70% rename from packages/nodes-base/nodes/Amqp/AmqpListener.node.ts rename to packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts index ab6fbbcf20..c65f9996e5 100644 --- a/packages/nodes-base/nodes/Amqp/AmqpListener.node.ts +++ b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts @@ -7,44 +7,27 @@ import { } from 'n8n-workflow'; -export class AmqpListener implements INodeType { +export class AmqpTrigger implements INodeType { description: INodeTypeDescription = { - displayName: 'AMQP Listener', - name: 'amqpListener', + displayName: 'AMQP Trigger', + name: 'amqpTrigger', icon: 'file:amqp.png', group: ['trigger'], version: 1, description: 'Listens to AMQP 1.0 Messages', defaults: { - name: 'AMQP Listener', + name: 'AMQP Trigger', color: '#00FF00', }, inputs: [], outputs: ['main'], credentials: [{ name: 'amqp', - required: false, + required: true, }], properties: [ // Node properties which the user gets displayed and // can change on the node. - { - displayName: 'Host', - name: 'hostname', - type: 'string', - default: 'localhost', - description: 'hostname of the amqp server', - }, - { - displayName: 'Port', - name: 'port', - type: 'number', - typeOptions: { - minValue: 1, - }, - default: 5672, - description: 'TCP Port to connect to', - }, { displayName: 'Queue / Topic', name: 'sink', @@ -76,9 +59,10 @@ export class AmqpListener implements INodeType { async trigger(this: ITriggerFunctions): Promise { const credentials = this.getCredentials('amqp'); + if (!credentials) { + throw new Error('Credentials are mandatory!'); + } - const hostname = this.getNodeParameter('hostname', 'localhost') as string; - const port = this.getNodeParameter('port', 5672) as number; const sink = this.getNodeParameter('sink', '') as string; const clientname = this.getNodeParameter('clientname', '') as string; const subscription = this.getNodeParameter('subscription', '') as string; @@ -88,19 +72,18 @@ export class AmqpListener implements INodeType { } let durable: boolean = false; if(subscription && clientname) { - console.log('durable subscription') durable = true; } let container = require('rhea'); let connectOptions = { - host: hostname, - port: port, + 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 container_id: (durable ? clientname : null) } - if (credentials) { + if (credentials.username || credentials.password) { container.options.username = credentials.username; container.options.password = credentials.password; } @@ -109,11 +92,8 @@ export class AmqpListener implements INodeType { let self = this; container.on('message', function (context: any) { - console.log('AMQP: received message id: ' + (context.message.message_id ? context.message.message_id : '')); - console.log(context.message.body); if (context.message.message_id && context.message.message_id == lastMsgId) { // ignore duplicate message check, don't think it's necessary, but it was in the rhea-lib example code - console.log('duplicate received: ' + context.message.message_id); lastMsgId = context.message.message_id; return; } @@ -141,25 +121,33 @@ export class AmqpListener implements INodeType { } } connection.open_receiver(clientOptions); - console.log('AMQP: listener attached'); // The "closeFunction" function gets called by n8n whenever // the workflow gets deactivated and can so clean up. async function closeFunction() { connection.close(); - console.log('AMQP: listener closed'); } // The "manualTriggerFunction" function gets called by n8n // when a user is in the workflow editor and starts the // workflow manually. - // does not make really sense for AMQP + // for AMQP it doesn't make much sense to wait here but + // for a new user who doesn't know how this works, it's better to wait and show a respective info message async function manualTriggerFunction() { - console.log('AMQP: manual trigger clicked, this will make the node spinn, until a message is received on: ' + sink); -/* self.emit([self.helpers.returnJsonArray([{ - error: '"manually triggered execution" stops the node right afterwards which unsubscribes the listener from the service bus. You need to activate the workflow to test.' - }])]); */ + + await new Promise( function( resolve ) { + let timeoutHandler = setTimeout(function() { + self.emit([self.helpers.returnJsonArray([{ + error: 'Aborted, no message received within 30secs. This 30sec timeout is only set for "manually triggered execution". Active Workflows will listen indefinitely.' + }])]); + resolve(true); + }, 30000); + container.on('message', function (context: any) { + clearTimeout(timeoutHandler); + resolve(true); + }); + }); } return { diff --git a/packages/nodes-base/package.json b/packages/nodes-base/package.json index 395409418a..d59d324207 100644 --- a/packages/nodes-base/package.json +++ b/packages/nodes-base/package.json @@ -60,7 +60,8 @@ "nodes": [ "dist/nodes/ActiveCampaign/ActiveCampaign.node.js", "dist/nodes/Airtable/Airtable.node.js", - "dist/nodes/Amqp/AmqpListener.node.js", + "dist/nodes/Amqp/Amqp.node.js", + "dist/nodes/Amqp/AmqpTrigger.node.js", "dist/nodes/Asana/Asana.node.js", "dist/nodes/Asana/AsanaTrigger.node.js", "dist/nodes/Aws/AwsLambda.node.js",