From 230dc33752f93d754df34f5ec5dc485267eeeda3 Mon Sep 17 00:00:00 2001 From: Leo Date: Fri, 1 Nov 2019 08:42:14 +0100 Subject: [PATCH 1/3] 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 0000000000..ec0c57014c --- /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 0000000000..ab6fbbcf20 --- /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/3] 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", From ab6cc43a4c1681c062abfe8f8e8105a669c25317 Mon Sep 17 00:00:00 2001 From: Jan Oberhauser Date: Sat, 2 Nov 2019 12:16:20 +0100 Subject: [PATCH 3/3] :shirt: Fix lint issues --- .../credentials/Amqp.credentials.ts | 3 -- packages/nodes-base/nodes/Amqp/Amqp.node.ts | 50 +++++++------------ .../nodes-base/nodes/Amqp/AmqpTrigger.node.ts | 41 +++++++-------- 3 files changed, 36 insertions(+), 58 deletions(-) diff --git a/packages/nodes-base/credentials/Amqp.credentials.ts b/packages/nodes-base/credentials/Amqp.credentials.ts index 640dedb18b..ff6b23ed1f 100644 --- a/packages/nodes-base/credentials/Amqp.credentials.ts +++ b/packages/nodes-base/credentials/Amqp.credentials.ts @@ -8,9 +8,6 @@ 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: 'Hostname', name: 'hostname', diff --git a/packages/nodes-base/nodes/Amqp/Amqp.node.ts b/packages/nodes-base/nodes/Amqp/Amqp.node.ts index 9dcb3ca5ef..66ed1aa25d 100644 --- a/packages/nodes-base/nodes/Amqp/Amqp.node.ts +++ b/packages/nodes-base/nodes/Amqp/Amqp.node.ts @@ -1,11 +1,11 @@ +import { ContainerOptions, Delivery } from 'rhea'; + import { IExecuteSingleFunctions } from 'n8n-core'; import { - IDataObject, INodeExecutionData, INodeType, INodeTypeDescription, } from 'n8n-workflow'; -import { Delivery } from 'rhea'; export class Amqp implements INodeType { description: INodeTypeDescription = { @@ -26,23 +26,6 @@ export class Amqp implements INodeType { 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', @@ -71,46 +54,47 @@ export class Amqp implements INodeType { } const sink = this.getNodeParameter('sink', '') as string; - let applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object; + const applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object; let headerProperties = applicationProperties; - if(typeof applicationProperties === 'string' && applicationProperties != '') { - headerProperties = JSON.parse(applicationProperties) + if(typeof applicationProperties === 'string' && applicationProperties !== '') { + headerProperties = JSON.parse(applicationProperties); } - if (sink == '') { + if (sink === '') { throw new Error('Queue or Topic required!'); } - let container = require('rhea'); + const container = require('rhea'); - let connectOptions = { + const connectOptions: ContainerOptions = { 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) { + const allSent = new Promise(( resolve ) => { + container.on('sendable', (context: any) => { // tslint:disable-line:no-any - let message = { + const message = { application_properties: headerProperties, body: JSON.stringify(item) - } - let sendResult = context.sender.send(message); + }; + + const 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 + const 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/AmqpTrigger.node.ts b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts index c65f9996e5..91a4eff552 100644 --- a/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts +++ b/packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts @@ -1,9 +1,10 @@ +import { ContainerOptions } from 'rhea'; + import { ITriggerFunctions } from 'n8n-core'; import { INodeType, INodeTypeDescription, ITriggerResponse, - } from 'n8n-workflow'; @@ -67,40 +68,40 @@ export class AmqpTrigger implements INodeType { const clientname = this.getNodeParameter('clientname', '') as string; const subscription = this.getNodeParameter('subscription', '') as string; - if (sink == '') { + if (sink === '') { throw new Error('Queue or Topic required!'); } - let durable: boolean = false; + let durable = false; if(subscription && clientname) { durable = true; } - let container = require('rhea'); - let connectOptions = { + const container = require('rhea'); + const connectOptions: ContainerOptions = { 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.username || credentials.password) { container.options.username = credentials.username; container.options.password = credentials.password; } - let lastMsgId: any = undefined; - let self = this; + let lastMsgId: number | undefined = undefined; + const self = this; - container.on('message', function (context: any) { - if (context.message.message_id && context.message.message_id == lastMsgId) { + container.on('message', (context: any) => { // tslint:disable-line:no-any + 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 lastMsgId = context.message.message_id; return; } self.emit([self.helpers.returnJsonArray([context.message])]); }); - - let connection = container.connect(connectOptions); + + const connection = container.connect(connectOptions); let clientOptions = undefined; if (durable) { clientOptions = { @@ -111,14 +112,14 @@ export class AmqpTrigger implements INodeType { expiry_policy: 'never' }, credit_window: 1 // prefetch 1 - } + }; } else { clientOptions = { source: { address: sink, }, credit_window: 1 // prefetch 1 - } + }; } connection.open_receiver(clientOptions); @@ -135,15 +136,11 @@ export class AmqpTrigger implements INodeType { // 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() { - - 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); + await new Promise(( resolve, reject ) => { + const timeoutHandler = setTimeout(() => { + 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', function (context: any) { + container.on('message', (context: any) => { // tslint:disable-line:no-any clearTimeout(timeoutHandler); resolve(true); });