n8n/packages/nodes-base/nodes/Amqp/Amqp.node.ts

126 lines
3.3 KiB
TypeScript
Raw Normal View History

2019-11-02 04:16:20 -07:00
import { ContainerOptions, Delivery } from 'rhea';
2019-11-01 05:50:33 -07:00
import { IExecuteSingleFunctions } from 'n8n-core';
import {
2020-08-06 05:42:26 -07:00
IDataObject,
2019-11-01 05:50:33 -07:00
INodeExecutionData,
INodeType,
INodeTypeDescription,
} from 'n8n-workflow';
2020-05-12 07:26:51 -07:00
export class Amqp implements INodeType {
2019-11-01 05:50:33 -07:00
description: INodeTypeDescription = {
displayName: 'AMQP Sender',
2020-05-12 07:26:51 -07:00
name: 'amqp',
2019-11-01 05:50:33 -07:00
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: '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.',
2020-08-06 05:42:26 -07:00
},
{
displayName: 'Options',
name: 'options',
type: 'collection',
placeholder: 'Add Option',
default: {},
options: [
{
displayName: 'Data as Object',
name: 'dataAsObject',
type: 'boolean',
default: false,
description: 'Send the data as an object.',
},
],
},
2019-11-01 05:50:33 -07:00
]
};
async executeSingle(this: IExecuteSingleFunctions): Promise<INodeExecutionData> {
const item = this.getInputData();
const credentials = this.getCredentials('amqp');
if (!credentials) {
throw new Error('Credentials are mandatory!');
}
const sink = this.getNodeParameter('sink', '') as string;
2019-11-02 04:16:20 -07:00
const applicationProperties = this.getNodeParameter('headerParametersJson', {}) as string | object;
2020-08-06 05:42:26 -07:00
const options = this.getNodeParameter('options', {}) as IDataObject;
2019-11-01 05:50:33 -07:00
let headerProperties = applicationProperties;
if (typeof applicationProperties === 'string' && applicationProperties !== '') {
2019-11-02 04:16:20 -07:00
headerProperties = JSON.parse(applicationProperties);
2019-11-01 05:50:33 -07:00
}
2019-11-02 04:16:20 -07:00
if (sink === '') {
2019-11-01 05:50:33 -07:00
throw new Error('Queue or Topic required!');
}
2019-11-02 04:16:20 -07:00
const container = require('rhea');
2019-11-01 05:50:33 -07:00
2019-11-02 04:16:20 -07:00
const connectOptions: ContainerOptions = {
2019-11-01 05:50:33 -07:00
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
2019-11-02 04:16:20 -07:00
};
2019-11-01 05:50:33 -07:00
if (credentials.username || credentials.password) {
container.options.username = credentials.username;
container.options.password = credentials.password;
}
2019-11-02 04:16:20 -07:00
const allSent = new Promise(( resolve ) => {
container.on('sendable', (context: any) => { // tslint:disable-line:no-any
2019-11-01 05:50:33 -07:00
2020-08-06 05:42:26 -07:00
let body: IDataObject | string = item.json;
if (options.dataAsObject !== true) {
body = JSON.stringify(body);
}
2019-11-02 04:16:20 -07:00
const message = {
2019-11-01 05:50:33 -07:00
application_properties: headerProperties,
2020-08-06 05:42:26 -07:00
body
2019-11-02 04:16:20 -07:00
};
2020-08-06 05:42:26 -07:00
2019-11-02 04:16:20 -07:00
const sendResult = context.sender.send(message);
2019-11-01 05:50:33 -07:00
resolve(sendResult);
});
});
2019-11-02 04:16:20 -07:00
2019-11-01 05:50:33 -07:00
container.connect(connectOptions).open_sender(sink);
2019-11-02 04:16:20 -07:00
const sendResult: Delivery = await allSent as Delivery; // sendResult has a a property that causes circular reference if returned
2019-11-01 05:50:33 -07:00
return { json: { id: sendResult.id } } as INodeExecutionData;
}
}