upgrade

Author: krystianity

CHANGELOG.md

@@ -1,5 +1,11 @@
 # kafka-streams CHANGELOG
 
+## 2019-07-04, Version 4.12.0
+
+* upgraded dependencies
+* fixed issue #91
+* added topic extractor example
+
 ## 2019-05-16, Version 4.11.0
 
 * added continueWith dsl operation

README.md

@@ -5,6 +5,7 @@
 [![npm version](https://badge.fury.io/js/kafka-streams.svg)](https://badge.fury.io/js/kafka-streams)
 
 ```
+// suggested Node.js version: v11.15.0
 npm install --save kafka-streams
 ```
 

examples/consumeFromTopic.js

@@ -25,5 +25,5 @@ stream.forEach((message) => {
 //(wait for the kafka consumer to be ready)
 stream.start().then(_ => {
     //wait a few ms and close all connections
-    setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 1000);
+    setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 5000);
 });

examples/produceToTopic.js

@@ -28,5 +28,5 @@ stream.start().then(_ => {
     ]);
 
     //wait a few ms and close all connections
-    setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 1000);
+    setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 5000);
 });

examples/topicExtractor.js

@@ -0,0 +1,49 @@
+"use strict";
+
+const { KafkaStreams } = require("./../index.js");
+const { nativeConfig: config } = require("./../test/test-config.js");
+
+const kafkaStreams = new KafkaStreams(config);
+
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
+const stream = kafkaStreams.getKStream(null);
+//creating a stream without topic is possible
+//no consumer will be created during stream.start()
+
+// 2nd arg is the output partition count, if you do not provide a partition the messages are automatically split across based on key or randomly
+// 3nd arg is the produce type of the sinek library, suggestion: stick to "send" this leaves all design options for you
+stream
+    .tap(console.log) // lets log some stream events
+    .to("default-target-topic", 1, "send");
+//define a topic to stream messages to, if nothing is actually defined in the kv structure
+
+//start the stream
+//(wait for the kafka producer to be ready, same as await .to())
+//and write a few messages to the topic stream.start()
+stream.start().then((_) => {
+
+    stream.writeToStream(getKafkaStyledMessage({ ping: "pong" })); // will be produced to default-target-topic
+    stream.writeToStream(getKafkaStyledMessage({ ping: "pong" }, "other-topic")); // will be produced to other-topic
+    stream.writeToStream([
+        getKafkaStyledMessage({ ping: "pong" }, "other-topic-2"),
+        getKafkaStyledMessage({ ping: "pong" }, "other-topic-3")
+    ]);
+
+    //wait a few ms and close all connections
+    setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 5000);
+});
+
+// its very important the events on your stream are shipped in a certain format
+// if they are not in the required KV format they will be treated as message values and not as full kafka messages
+function getKafkaStyledMessage(payload, topic = undefined, partition = undefined) {
+    return {
+        key: null, // (required) just to not be undefined, keys will otherwise receive random uuids
+        value: JSON.stringify(payload), // required, ensure this is a string or a buffer!
+        // optional:
+        topic,
+        partition,
+    };
+}

examples/window.js

@@ -32,4 +32,4 @@ stream
 //start the stream
 consumeStream.start();
 
-//setTimeout(abort, 5000); // -> abort the window collection after 5 seconds
+//setTimeout(() => { abort(); }, 5000); // -> abort the window collection after 5 seconds

lib/messageProduceHandle.js

@@ -142,10 +142,14 @@ const messageProduceHandle = (kafka, message, outputTopicName, produceType, comp
         _version
 
     ).then((produceMessageValue) => {
+
+        debug("Produce successfull", kafkaMessage.topic, produceMessageValue);
         if (produceHandler) {
             produceHandler.emit("delivered", kafkaMessage, produceMessageValue);
         }
     }).catch((error) => {
+
+        debug("Produce failed", kafkaMessage, error.message);
         if (producerErrorCallback) {
             error.message = "During message produce: " + error.message;
             producerErrorCallback(error);

package.json

@@ -1,6 +1,6 @@
 {
     "name": "kafka-streams",
-    "version": "4.11.0",
+    "version": "4.12.0",
     "description": "kafka-streams for Node.js",
     "typings": "index.d.ts",
     "main": "index.js",
@@ -15,6 +15,7 @@
         "window": "node ./examples/window.js",
         "wordCount": "node ./examples/wordCount.js",
         "word-count": "yarn wordCount",
+        "topicExtractor": "DEBUG=kafka-streams:mph node ./examples/topicExtractor.js",
         "obs-test": "_mocha test/unit/Observable.test.js",
         "test-unit": "_mocha --recursive --exit -R spec test/unit",
         "test-int": "_mocha --recursive --timeout 22500 --exit -R spec test/int",
@@ -59,20 +60,20 @@
     },
     "homepage": "https://github.com/nodefluent/kafka-streams#readme",
     "dependencies": {
-        "bluebird": "~3.5.4",
+        "bluebird": "~3.5.5",
         "debug": "~4.1.1",
-        "global": "^4.3.2",
-        "lodash.clone": "^4.5.0",
-        "lodash.clonedeep": "^4.5.0",
-        "most": "~1.7.3",
-        "most-subject": "~5.3.0",
-        "sinek": "~7.29.3"
+        "global": "~4.4.0",
+        "lodash.clone": "~4.5.0",
+        "lodash.clonedeep": "~4.5.0",
+        "most": "1.7.3",
+        "most-subject": "5.3.0",
+        "sinek": "~7.30.0"
     },
     "devDependencies": {
-        "async": "~2.6.2",
-        "jsdoc": "~3.6.1",
-        "jsdoc-to-markdown": "~4.0.1",
-        "log4bro": "~3.9.0",
+        "async": "~3.1.0",
+        "jsdoc": "~3.6.2",
+        "jsdoc-to-markdown": "~5.0.0",
+        "log4bro": "~3.10.0",
         "mocha": "~6.1.4",
         "proxyquire": "~2.1.0",
         "uuid": "~3.3.2"