mohiit1502/Automata
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

⚡ Some improvements on AMQP nodes

Browse Source
This commit is contained in:
Jan Oberhauser
2021-01-20 13:20:48 +01:00
parent 64e36f04b0
commit 8432a8774e
2 changed files with 92 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
packages/nodes-base/nodes/Amqp/Amqp.node.ts
View File

@@ -1,5 +1,9 @@
import { ContainerOptions, Dictionary, EventContext } from 'rhea';
import rhea = require('rhea');
import {
create_container,
ContainerOptions,
Dictionary,
EventContext,
} from 'rhea';
import { IExecuteFunctions } from 'n8n-core';
import {
@@ -51,6 +55,13 @@ export class Amqp implements INodeType {
placeholder: 'Add Option',
default: {},
options: [
{
displayName: 'Container ID',
name: 'containerId',
type: 'string',
default: '',
description: 'Will be used to pass to the RHEA Backend as container_id',
},
{
displayName: 'Data as Object',
name: 'dataAsObject',
@@ -58,6 +69,20 @@ export class Amqp implements INodeType {
default: false,
description: 'Send the data as an object.',
},
{
displayName: 'Reconnect',
name: 'reconnect',
type: 'boolean',
default: true,
description: 'Automatically reconnect if disconnected',
},
{
displayName: 'Reconnect Limit',
name: 'reconnectLimit',
type: 'number',
default: 50,
description: 'Maximum number of reconnect attempts',
},
{
displayName: 'Send property',
name: 'sendOnlyProperty',
@@ -65,33 +90,12 @@ export class Amqp implements INodeType {
default: '',
description: 'The only property to send. If empty the whole item will be sent.',
},
{
displayName: 'Container ID',
name: 'containerID',
type: 'string',
default: '',
description: 'Will be used to pass to the RHEA Backend as container_id',
},
{
displayName: 'Reconnect',
name: 'reconnect',
type: 'boolean',
default: true,
description: 'If on, the library will automatically attempt to reconnect if disconnected',
},
{
displayName: 'Reconnect limit',
name: 'reconnectLimit',
type: 'number',
default: 50,
description: 'maximum number of reconnect attempts',
},
],
},
],
};
async execute(this: IExecuteFunctions): Promise < INodeExecutionData[][] > {
async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
const credentials = this.getCredentials('amqp');
if (!credentials) {
throw new Error('Credentials are mandatory!');
@@ -100,22 +104,22 @@ export class Amqp implements INodeType {
const sink = this.getNodeParameter('sink', 0, '') as string;
const applicationProperties = this.getNodeParameter('headerParametersJson', 0, {}) as string | object;
const options = this.getNodeParameter('options', 0, {}) as IDataObject;
const container_id = options.containerID as string;
const containerReconnect = options.reconnect as boolean || true ;
const containerId = options.containerId as string;
const containerReconnect = options.reconnect as boolean || true;
const containerReconnectLimit = options.reconnectLimit as number || 50;
let headerProperties : Dictionary<any>;
let headerProperties: Dictionary<any>; // tslint:disable-line:no-any
if (typeof applicationProperties === 'string' && applicationProperties !== '') {
headerProperties = JSON.parse(applicationProperties);
} else {
headerProperties = applicationProperties as object;
}
}
if (sink === '') {
throw new Error('Queue or Topic required!');
}
const container = rhea.create_container();
const container = create_container();
/*
Values are documentet here: https://github.com/amqp/rhea#container
@@ -124,20 +128,20 @@ export class Amqp implements INodeType {
host: credentials.hostname,
hostname: credentials.hostname,
port: credentials.port,
reconnect: containerReconnect,
reconnect: containerReconnect,
reconnect_limit: containerReconnectLimit,
username: credentials.username ? credentials.username : undefined,
password: credentials.password ? credentials.password : undefined,
transport: credentials.transportType ? credentials.transportType : undefined,
container_id: container_id ? container_id : undefined,
id: container_id ? container_id : undefined,
container_id: containerId ? containerId : undefined,
id: containerId ? containerId : undefined,
};
const conn = container.connect(connectOptions);
const sender = conn.open_sender(sink);
const responseData: IDataObject[] = await new Promise((resolve) => {
container.once('sendable', (context: EventContext) => {
container.once('sendable', (context: EventContext) => {
const returnData = [];
const items = this.getInputData();