mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-12-25 04:34:06 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

⚡ Improvements to Amqp Node

Browse source
This commit is contained in:
Jan Oberhauser 2020-10-28 23:30:30 +01:00
parent 6f92ecb128
commit 708b0c8b26
3 changed files with 27 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
packages/nodes-base/credentials/Amqp.credentials.ts
View file

@ -41,6 +41,7 @@ export class Amqp implements ICredentialType {
name: 'transportType',
type: 'string' as NodePropertyTypes,
default: '',
description: 'Optional Transport Type to use.',
},
];
}

15
packages/nodes-base/nodes/Amqp/Amqp.node.ts
View file

@ -62,11 +62,11 @@ export class Amqp implements INodeType {
name: 'sendOnlyProperty',
type: 'string',
default: '',
description: 'Send only this property - If empty the hole Json will be sent',
description: 'The only property to send. If empty the whole item will be sent.',
},
],
},
]
],
};
async executeSingle(this: IExecuteSingleFunctions): Promise<INodeExecutionData> {
@ -105,7 +105,7 @@ export class Amqp implements INodeType {
connectOptions.username = credentials.username;
connectOptions.password = credentials.password;
}
if (credentials.transportType) {
if (credentials.transportType !== '') {
connectOptions.transport = credentials.transportType;
}
@ -113,11 +113,10 @@ export class Amqp implements INodeType {
container.on('sendable', (context: any) => { // tslint:disable-line:no-any
let body: IDataObject | string = item.json;
let prop = options.sendOnlyProperty as string;
const sendOnlyProperty = options.sendOnlyProperty as string;
if(prop)
{
body = body[prop] as string;
if (sendOnlyProperty) {
body = body[sendOnlyProperty] as string;
}
if (options.dataAsObject !== true) {
@ -126,7 +125,7 @@ export class Amqp implements INodeType {
const message = {
application_properties: headerProperties,
body
body,
};
const sendResult = context.sender.send(message);

39
packages/nodes-base/nodes/Amqp/AmqpTrigger.node.ts
View file

@ -62,11 +62,11 @@ export class AmqpTrigger implements INodeType {
default: {},
options: [
{
displayName: 'Only Body',
name: 'onlyBody',
displayName: 'Convert Body To String',
name: 'jsonConvertByteArrayToString',
type: 'boolean',
default: false,
description: 'Returns only the body property.',
description: 'Convert JSON Body content (["body"]["content"]) from Byte Array to string. Needed for Azure Service Bus.',
},
{
displayName: 'JSON Parse Body',
@ -76,15 +76,15 @@ export class AmqpTrigger implements INodeType {
description: 'Parse the body to an object.',
},
{
displayName: 'Convert JSON Body content from Byte Array to string',
name: 'jsonConvertByteArrayToString',
displayName: 'Only Body',
name: 'onlyBody',
type: 'boolean',
default: false,
description: 'Convert JSON Body content (["body"]["content"]) from Byte Array to string - Azure Service Bus',
}
description: 'Returns only the body property.',
},
],
},
]
],
};
@ -117,7 +117,7 @@ export class AmqpTrigger implements INodeType {
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)
container_id: (durable ? clientname : null),
};
if (credentials.username || credentials.password) {
// Old rhea implementation. not shure if it is neccessary
@ -125,7 +125,7 @@ export class AmqpTrigger implements INodeType {
container.options.password = credentials.password;
connectOptions.username = credentials.username;
connectOptions.password = credentials.password;
}
}
if (credentials.transportType) {
connectOptions.transport = credentials.transportType;
}
@ -139,15 +139,14 @@ export class AmqpTrigger implements INodeType {
if (context.message.message_id && context.message.message_id === lastMsgId) {
return;
}
console.log("new Message", context.message.message_id, lastMsgId);
lastMsgId = context.message.message_id;
let data = context.message;
if(options.jsonConvertByteArrayToString === true && data.body.content !== undefined) {
if (options.jsonConvertByteArrayToString === true && data.body.content !== undefined) {
// The buffer is not ready... Stringify and parse back to load it.
let cont = JSON.stringify(data.body.content);
data.body = String.fromCharCode.apply(null,JSON.parse(cont).data);
const content = JSON.stringify(data.body.content);
data.body = String.fromCharCode.apply(null, JSON.parse(content).data);
}
if (options.jsonParseBody === true) {
@ -169,16 +168,16 @@ export class AmqpTrigger implements INodeType {
source: {
address: sink,
durable: 2,
expiry_policy: 'never'
expiry_policy: 'never',
},
credit_window: 1 // prefetch 1
credit_window: 1, // prefetch 1
};
} else {
clientOptions = {
source: {
address: sink,
},
credit_window: 1 // prefetch 1
credit_window: 1, // prefetch 1
};
}
connection.open_receiver(clientOptions);
@ -196,10 +195,10 @@ 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(( resolve, reject ) => {
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.'));
}, 3000);
}, 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