kafkajs-stream - npm Package Compare versions

Comparing version 0.0.3 to 0.0.4

CHANGELOG.md

@@ -5,2 +5,9 @@ # Changelog
 
+### [0.0.4](https://github.com/kambing86/kafkajs-stream/compare/v0.0.3...v0.0.4) (2019-12-13)
+
+
+### Bug Fixes
+
+* use paused flag instead of error ([82b341b](https://github.com/kambing86/kafkajs-stream/commit/82b341bbdc70342d4aa98f3536be53ae3fd54696))
+
 ### [0.0.3](https://github.com/kambing86/kafkajs-stream/compare/v0.0.2...v0.0.3) (2019-12-13)
@@ -7,0 +14,0 @@

dist/ConsumerStream.js

@@ -13,4 +13,2 @@ "use strict";
 const stream_1 = require("stream");
-class TooMuchError extends Error {
-}
 class ConsumerStream extends stream_1.Readable {
@@ -39,9 +37,2 @@ constructor(kafka, options) {
                     this.paused = false;
-                    try {
-                        this.consumer.resume([{ topic: this.topic.topic }]);
-                    }
-                    catch (e) {
-                        // consumer might be stopped for some reasons, and calling resume will throw error
-                        yield this.run();
-                    }
                 }
@@ -57,4 +48,11 @@ }
             yield this.consumer.run({
+                eachBatchAutoResolve: false,
                 eachBatch: ({ batch, resolveOffset, heartbeat }) => __awaiter(this, void 0, void 0, function* () {
+                    if (this.paused) {
+                        return;
+                    }
                     for (const message of batch.messages) {
+                        if (this.paused) {
+                            break;
+                        }
                         const continueToPush = this.push(message.value);
@@ -64,3 +62,3 @@ resolveOffset(message.offset);
                         if (!continueToPush) {
-                            throw new TooMuchError();
+                            this.paused = true;
                         }
@@ -67,0 +65,0 @@ }

package.json

 {
   "name": "kafkajs-stream",
-  "version": "0.0.3",
+  "version": "0.0.4",
   "description": "Stream for kafkajs",
@@ -5,0 +5,0 @@ "main": "dist/index.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

14298

increased by0.14%

Worsened metrics

Lines of code

174

decreased by-1.14%

No dependency changes