mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-03-05 20:50:17 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix Subscription

Browse source
This commit is contained in:
Günther Erb 2020-12-30 13:57:53 +01:00
parent 7648ee8d93
commit 1a1accce43
1 changed files with 10 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,4 +1,4 @@
import { Container, ContainerOptions, EventContext, ReceiverOptions } from 'rhea'; import { Container, ContainerOptions, EventContext, Message, ReceiverOptions } from 'rhea';
import { ITriggerFunctions } from 'n8n-core'; import { ITriggerFunctions } from 'n8n-core';
import { import {
@ -169,11 +169,11 @@ export class AmqpTrigger implements INodeType {
let lastMsgId: string| number | Buffer | undefined = undefined; let lastMsgId: string| number | Buffer | undefined = undefined;
const self = this; const self = this;
container.on('receiver_open', (context: any) => { // tslint:disable-line:no-any container.on('receiver_open', (context: EventContext) => {
context.receiver.add_credit(pullMessagesNumber); context.receiver?.add_credit(pullMessagesNumber);
}); });
container.on('message', (context: EventContext) => { // tslint:disable-line:no-any container.on('message', (context: EventContext) => {
// No message in the context // No message in the context
if(!context.message) if(!context.message)
@ -241,6 +241,7 @@ export class AmqpTrigger implements INodeType {
const connection = container.connect(connectOptions); const connection = container.connect(connectOptions);
let clientOptions : ReceiverOptions = { let clientOptions : ReceiverOptions = {
name: subscription ? subscription : undefined,
source: { source: {
address: sink, address: sink,
durable: (durable ? 2 : undefined), durable: (durable ? 2 : undefined),
@ -269,14 +270,15 @@ export class AmqpTrigger implements INodeType {
const timeoutHandler = setTimeout(() => { 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.')); 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); }, 3000);
container.on('message', (context: any) => { // tslint:disable-line:no-any container.on('message', (context: EventContext) => {
// Check if the only property present in the message is body // Check if the only property present in the message is body
// in which case we only emit the content of the body property // in which case we only emit the content of the body property
// otherwise we emit all properties and their content // otherwise we emit all properties and their content
if (Object.keys(context.message)[0] === 'body' && Object.keys(context.message).length === 1) { const message = context.message as Message;
self.emit([self.helpers.returnJsonArray([context.message.body])]); if (Object.keys(message)[0] === 'body' && Object.keys(message).length === 1) {
self.emit([self.helpers.returnJsonArray([message.body])]);
} else { } else {
self.emit([self.helpers.returnJsonArray([context.message])]); self.emit([self.helpers.returnJsonArray([message as any])]);
} }
clearTimeout(timeoutHandler); clearTimeout(timeoutHandler);
resolve(true); resolve(true);