diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..550c62e --- /dev/null +++ b/.env.example @@ -0,0 +1,4 @@ +export CLOUDKARAFKA_BROKERS="host1:9094,host2:9094,host3:9094" +export CLOUDKARAFKA_USERNAME="username" +export CLOUDKARAFKA_PASSWORD="password" +export CLOUDKARAFKA_TOPIC="default" \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5702678 --- /dev/null +++ b/.gitignore @@ -0,0 +1,65 @@ + +# Created by https://www.gitignore.io/api/node + +### Node ### +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* + +# Runtime data +pids +*.pid +*.seed +*.pid.lock + +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov + +# Coverage directory used by tools like istanbul +coverage + +# nyc test coverage +.nyc_output + +# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files) +.grunt + +# Bower dependency directory (https://bower.io/) +bower_components + +# node-waf configuration +.lock-wscript + +# Compiled binary addons (http://nodejs.org/api/addons.html) +build/Release + +# Dependency directories +node_modules/ +jspm_packages/ + +# Typescript v1 declaration files +typings/ + +# Optional npm cache directory +.npm + +# Optional eslint cache +.eslintcache + +# Optional REPL history +.node_repl_history + +# Output of 'npm pack' +*.tgz + +# Yarn Integrity file +.yarn-integrity + +# dotenv environment variables file +.env + + +# End of https://www.gitignore.io/api/node diff --git a/README.md b/README.md index 2f31531..b0a17df 100644 --- a/README.md +++ b/README.md @@ -12,21 +12,24 @@ Configuration Username can be found in the Details view in for your CloudKarafka instance. * `export CLOUDKARAFKA_PASSWORD="password"` Password can be found in the Details view in for your CloudKarafka instance. -* `export CLOUDKARAFKA_TOPIC_PREFIX="-"` - Topic prefix should be the same as your username. +* `export CLOUDKARAFKA_TOPIC="default"` + Topic name without the "prefix-" that is your username. -These export commands must be run in both of the terminal windows below. +These exports must be run in both of the terminal windows below, or create a `.env` file according to the example then follow the commands. ``` git clone https://github.com/CloudKarafka/nodejs-kafka-example.git cd nodejs-kafka-example npm install +cp .env.example .env +# now add your own credentials from the dashboard to the `.env` file +. ./.env node consumer.js ``` Open another terminal window ``` -# setup ENV variables for the new terminal window (from above) cd nodejs-kafka-example -node consumer.js +. ./.env +node producer.js ``` diff --git a/consumer.js b/consumer.js index 6da9a54..7c90205 100644 --- a/consumer.js +++ b/consumer.js @@ -1,7 +1,7 @@ var Kafka = require("node-rdkafka"); var kafkaConf = { - "group.id": "cloudkarafka-example", + "group.id": process.env.CLOUDKARAFKA_USERNAME + "-consumer", "metadata.broker.list": process.env.CLOUDKARAFKA_BROKERS.split(","), "socket.keepalive.enable": true, "security.protocol": "SASL_SSL", @@ -12,21 +12,21 @@ var kafkaConf = { }; const prefix = process.env.CLOUDKARAFKA_USERNAME; -const topics = [`${prefix}.test`]; +const topics = [`${prefix}-${process.env.CLOUDKARAFKA_TOPIC}`]; const consumer = new Kafka.KafkaConsumer(kafkaConf, { "auto.offset.reset": "beginning" }); const numMessages = 5; let counter = 0; -consumer.on("error", function(err) { +consumer.on("error", function (err) { console.error(err); }); -consumer.on("ready", function(arg) { +consumer.on("ready", function (arg) { console.log(`Consumer ${arg.name} ready`); consumer.subscribe(topics); consumer.consume(); }); -consumer.on("data", function(m) { +consumer.on("data", function (m) { counter++; if (counter % numMessages === 0) { console.log("calling commit"); @@ -34,18 +34,18 @@ consumer.on("data", function(m) { } console.log(m.value.toString()); }); -consumer.on("disconnected", function(arg) { +consumer.on("disconnected", function (arg) { process.exit(); }); -consumer.on('event.error', function(err) { +consumer.on('event.error', function (err) { console.error(err); process.exit(1); }); -consumer.on('event.log', function(log) { +consumer.on('event.log', function (log) { console.log(log); }); consumer.connect(); -setTimeout(function() { +setTimeout(function () { consumer.disconnect(); }, 300000); diff --git a/producer.js b/producer.js index bb550fd..f66c9eb 100644 --- a/producer.js +++ b/producer.js @@ -1,7 +1,7 @@ const Kafka = require("node-rdkafka"); const kafkaConf = { - "group.id": "cloudkarafka-example", + "group.id": process.env.CLOUDKARAFKA_USERNAME + "-consumer", "metadata.broker.list": process.env.CLOUDKARAFKA_BROKERS.split(","), "socket.keepalive.enable": true, "security.protocol": "SASL_SSL", @@ -11,14 +11,14 @@ const kafkaConf = { "debug": "generic,broker,security" }; -const prefix = process.env.CLOUDKARAFKA_TOPIC_PREFIX; -const topic = `${prefix}.test`; +const prefix = process.env.CLOUDKARAFKA_USERNAME; +const topic = `${prefix}-${process.env.CLOUDKARAFKA_TOPIC}`; const producer = new Kafka.Producer(kafkaConf); const maxMessages = 20; const genMessage = i => new Buffer(`Kafka example, message number ${i}`); -producer.on("ready", function(arg) { +producer.on("ready", function (arg) { console.log(`producer ${arg.name} ready.`); for (var i = 0; i < maxMessages; i++) { producer.produce(topic, -1, genMessage(i), i); @@ -26,15 +26,15 @@ producer.on("ready", function(arg) { setTimeout(() => producer.disconnect(), 0); }); -producer.on("disconnected", function(arg) { +producer.on("disconnected", function (arg) { process.exit(); }); -producer.on('event.error', function(err) { +producer.on('event.error', function (err) { console.error(err); process.exit(1); }); -producer.on('event.log', function(log) { +producer.on('event.log', function (log) { console.log(log); }); producer.connect();