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

⚡ Allow fromBeginning config in kafka trigger node (#1958)

Browse source 
* Allow fromBeginning config in kafka trigger node

* make sure options in defined
This commit is contained in:
Pierre Lanvin 2021-07-14 20:31:43 +02:00 committed by GitHub
parent c983603306
commit 58f0d7cffc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -70,6 +70,13 @@ export class KafkaTrigger implements INodeType {
default: false,
description: 'Allow sending message to a previously non exisiting topic .',
},
{
displayName: 'Read messages from beginning',
name: 'fromBeginning',
type: 'boolean',
default: true,
description: 'Read message from beginning .',
},
{
displayName: 'JSON Parse Message',
name: 'jsonParseMessage',
@ -140,13 +147,13 @@ export class KafkaTrigger implements INodeType {
const consumer = kafka.consumer({ groupId });
await consumer.connect();
const options = this.getNodeParameter('options', {}) as IDataObject;
await consumer.subscribe({ topic, fromBeginning: true });
await consumer.subscribe({ topic, fromBeginning: (options.fromBeginning)? true : false });
const self = this;
const options = this.getNodeParameter('options', {}) as IDataObject;
const startConsumer = async () => {
await consumer.run({
eachMessage: async ({ topic, message }) => {