From 230dc33752f93d754df34f5ec5dc485267eeeda3 Mon Sep 17 00:00:00 2001 From: Leo Date: Fri, 1 Nov 2019 08:42:14 +0100 Subject: [PATCH 1/2] amqp listener with debug-console output --- .../credentials/Amqp.credentials.ts | 30 +++ .../nodes/Amqp/AmqpListener.node.ts | 171 ++++++++++++++++++ packages/nodes-base/nodes/Amqp/amqp.png | Bin 0 -> 550 bytes packages/nodes-base/package.json | 3 + 4 files changed, 204 insertions(+) create mode 100644 packages/nodes-base/credentials/Amqp.credentials.ts create mode 100644 packages/nodes-base/nodes/Amqp/AmqpListener.node.ts create mode 100644 packages/nodes-base/nodes/Amqp/amqp.png diff --git a/packages/nodes-base/credentials/Amqp.credentials.ts b/packages/nodes-base/credentials/Amqp.credentials.ts new file mode 100644 index 000000000..ec0c57014 --- /dev/null +++ b/packages/nodes-base/credentials/Amqp.credentials.ts @@ -0,0 +1,30 @@ +import { + ICredentialType, + NodePropertyTypes, +} from 'n8n-workflow'; + + +export class Amqp implements ICredentialType { + name = 'amqp'; + displayName = 'AMQP'; + properties = [ + // The credentials to get from user and save encrypted. + // Properties can be defined exactly in the same way + // as node properties. + { + displayName: 'User', + name: 'username', + type: 'string' as NodePropertyTypes, + default: '', + }, + { + displayName: 'Password', + name: 'password', + type: 'string' as NodePropertyTypes, + typeOptions: { + password: true, + }, + default: '', + }, + ]; +} diff --git a/packages/nodes-base/nodes/Amqp/AmqpListener.node.ts b/packages/nodes-base/nodes/Amqp/AmqpListener.node.ts new file mode 100644 index 000000000..ab6fbbcf2 --- /dev/null +++ b/packages/nodes-base/nodes/Amqp/AmqpListener.node.ts @@ -0,0 +1,171 @@ +import { ITriggerFunctions } from 'n8n-core'; +import { + INodeType, + INodeTypeDescription, + ITriggerResponse, + +} from 'n8n-workflow'; + + +export class AmqpListener implements INodeType { + description: INodeTypeDescription = { + displayName: 'AMQP Listener', + name: 'amqpListener', + icon: 'file:amqp.png', + group: ['trigger'], + version: 1, + description: 'Listens to AMQP 1.0 Messages', + defaults: { + name: 'AMQP Listener', + color: '#00FF00', + }, + inputs: [], + outputs: ['main'], + credentials: [{ + name: 'amqp', + required: false, + }], + 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', + type: 'string', + default: '', + placeholder: 'topic://sourcename.something', + description: 'name of the queue of topic to listen to', + }, + { + displayName: 'Clientname', + name: 'clientname', + type: 'string', + default: '', + placeholder: 'for durable/persistent topic subscriptions, example: "n8n"', + description: 'Leave empty for non-durable topic subscriptions or queues. ', + }, + { + displayName: 'Subscription', + name: 'subscription', + type: 'string', + default: '', + placeholder: 'for durable/persistent topic subscriptions, example: "order-worker"', + description: 'Leave empty for non-durable topic subscriptions or queues', + }, + ] + }; + + + async trigger(this: ITriggerFunctions): Promise { + + const credentials = this.getCredentials('amqp'); + + 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; + + if (sink == '') { + throw new Error('Queue or Topic required!'); + } + let durable: boolean = false; + if(subscription && clientname) { + console.log('durable subscription') + durable = true; + } + + let container = require('rhea'); + let connectOptions = { + host: hostname, + port: 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) { + container.options.username = credentials.username; + container.options.password = credentials.password; + } + + let lastMsgId: any = undefined; + 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; + } + self.emit([self.helpers.returnJsonArray([context.message])]); + }); + + let connection = container.connect(connectOptions); + let clientOptions = undefined; + if (durable) { + clientOptions = { + name: subscription, + source: { + address: sink, + durable: 2, + expiry_policy: 'never' + }, + credit_window: 1 // prefetch 1 + } + } else { + clientOptions = { + source: { + address: sink, + }, + credit_window: 1 // prefetch 1 + } + } + 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 + 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.' + }])]); */ + } + + return { + closeFunction, + manualTriggerFunction, + }; + + } +} diff --git a/packages/nodes-base/nodes/Amqp/amqp.png b/packages/nodes-base/nodes/Amqp/amqp.png new file mode 100644 index 0000000000000000000000000000000000000000..0f0345078c91652ec2d8cbdc5a600b3f8d2f06ce GIT binary patch literal 550 zcmeAS@N?(olHy`uVBq!ia0vp^HXzKw1|+Ti+$>^XV0`Q8;uws1f$ymQRBRD& z;(!tSPRVKaFI;Pk&fc_pEAP#czOyznUOSbs8%JAidn3Pn@=V=jEk*9aF=`xk2fkN6 z5a3!TyYSAf&Zc5*p;*@4k9fbalz*r-RNA2>F#Gm9zJux|-!ujGc=Z`1U;M=+So$mW zvU$Glowti;+Rnat=FGhencGV8nP<9Z#xC6Nw%EgfYj@CoagGwR$+D-p=QSJh{*7-s zal7U=<9(ZTPu%(a3w#4Bm%qQqI?1wm^Pa;K6JlP!Nnx_K7b);=Jf`sN*=Lqymi?ES z=3g*Zk&w-qG~-QziUey#=K0!*4|rqN)wujiI4H__%evX3$w%ziOrfQcTuz3?%lqGY zzmVk=JCk1W_D-K!QD0x*-07<%_a&?nx!_&ZRk1ec;Py2&vM1KW-#WB{61hEZ1 z?{D$kGf#Q5&FxUS5@&=MSEP@wc>4>DR9n9(Zlx10l?J@zxq2xyz<>J4(tJi2JOS)PB_ Date: Fri, 1 Nov 2019 13:50:33 +0100 Subject: [PATCH 2/2] 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 ec0c57014..640dedb18 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 000000000..9dcb3ca5e --- /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 ab6fbbcf2..c65f9996e 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 395409418..d59d32420 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",