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

feat(Kafka Trigger Node): Add non-parallel execution (#6175)

Browse source 
* Fix typo, add v1.1

* Add parallel processing

* Add versioning

* Improve description for maximum inflight requests

---------

Co-authored-by: Michael Kret <michael.k@radency.com>
This commit is contained in:
agobrech 2023-05-04 17:26:56 +02:00 committed by GitHub
parent 0eb4d9fc16
commit 814ea5185c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 33 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts
View file

@ -9,8 +9,9 @@ import type {
INodeType, INodeType,
INodeTypeDescription, INodeTypeDescription,
ITriggerResponse, ITriggerResponse,
IRun,
} from 'n8n-workflow'; } from 'n8n-workflow';
import { NodeOperationError } from 'n8n-workflow'; import { createDeferredPromise, NodeOperationError } from 'n8n-workflow';
export class KafkaTrigger implements INodeType { export class KafkaTrigger implements INodeType {
description: INodeTypeDescription = { description: INodeTypeDescription = {
@ -18,7 +19,7 @@ export class KafkaTrigger implements INodeType {
name: 'kafkaTrigger', name: 'kafkaTrigger',
icon: 'file:kafka.svg', icon: 'file:kafka.svg',
group: ['trigger'], group: ['trigger'],
version: 1, version: [1, 1.1],
description: 'Consume messages from a Kafka topic', description: 'Consume messages from a Kafka topic',
defaults: { defaults: {
name: 'Kafka Trigger', name: 'Kafka Trigger',
@ -116,7 +117,7 @@ export class KafkaTrigger implements INodeType {
type: 'number', type: 'number',
default: 1, default: 1,
description: description:
'Max number of requests that may be in progress at any time. If falsey then no limit.', 'The maximum number of unacknowledged requests the client will send on a single connection',
}, },
{ {
displayName: 'Read Messages From Beginning', displayName: 'Read Messages From Beginning',
@ -132,6 +133,19 @@ export class KafkaTrigger implements INodeType {
default: false, default: false,
description: 'Whether to try to parse the message to an object', description: 'Whether to try to parse the message to an object',
}, },
{
displayName: 'Parallel Processing',
name: 'parallelProcessing',
type: 'boolean',
default: true,
displayOptions: {
hide: {
'@version': [1],
},
},
description:
'Whether to process messages in parallel or by keeping the message in order',
},
{ {
displayName: 'Only Message', displayName: 'Only Message',
name: 'onlyMessage', name: 'onlyMessage',
@ -177,6 +191,10 @@ export class KafkaTrigger implements INodeType {
const ssl = credentials.ssl as boolean; const ssl = credentials.ssl as boolean;
const options = this.getNodeParameter('options', {}) as IDataObject;
options.nodeVersion = this.getNode().typeVersion;
const config: KafkaConfig = { const config: KafkaConfig = {
clientId, clientId,
brokers, brokers,
@ -213,9 +231,9 @@ export class KafkaTrigger implements INodeType {
heartbeatInterval: this.getNodeParameter('options.heartbeatInterval', 3000) as number, heartbeatInterval: this.getNodeParameter('options.heartbeatInterval', 3000) as number,
}); });
await consumer.connect(); const parallelProcessing = options.parallelProcessing as boolean;
const options = this.getNodeParameter('options', {}) as IDataObject; await consumer.connect();
await consumer.subscribe({ topic, fromBeginning: options.fromBeginning ? true : false }); await consumer.subscribe({ topic, fromBeginning: options.fromBeginning ? true : false });
@ -261,8 +279,16 @@ export class KafkaTrigger implements INodeType {
//@ts-ignore //@ts-ignore
data = value; data = value;
} }
let responsePromise = undefined;
if (!parallelProcessing && (options.nodeVersion as number) > 1) {
responsePromise = await createDeferredPromise<IRun>();
this.emit([this.helpers.returnJsonArray([data])], undefined, responsePromise);
} else {
this.emit([this.helpers.returnJsonArray([data])]); this.emit([this.helpers.returnJsonArray([data])]);
}
if (responsePromise) {
await responsePromise.promise();
}
}, },
}); });
}; };