kafkajs - npm Package Compare versions

Comparing version 1.0.0 to 1.0.1

CHANGELOG.md

@@ -8,2 +8,6 @@ # Changelog
 
+## [1.0.1] - 2018-05-18
+### Fixed
+  - Prevent crash when re-producing after metadata refresh #62
+
 ## [1.0.0] - 2018-05-14
@@ -10,0 +14,0 @@ ## Changed

package.json

 {
   "name": "kafkajs",
-  "version": "1.0.0",
+  "version": "1.0.1",
   "description": "A modern Apache Kafka client for node.js",
@@ -50,3 +50,3 @@ "author": "Tulio Ornelas <[e-mail]>",
   "dependencies": {
-    "long": "^3.2.0"
+    "long": "^4.0.0"
   },
@@ -53,0 +53,0 @@ "lint-staged": {

src/producer/sendMessages.js

@@ -48,4 +48,9 @@ const createRetry = require('../retry')
 
-        const response = await broker.produce({ acks, timeout, compression, topicData })
-        responsePerBroker.set(broker, responseSerializer(response))
+        try {
+          const response = await broker.produce({ acks, timeout, compression, topicData })
+          responsePerBroker.set(broker, responseSerializer(response))
+        } catch (e) {
+          responsePerBroker.delete(broker)
+          throw e
+        }
       })
@@ -52,0 +57,0 @@ }

src/protocol/decoder.js

@@ -8,2 +8,5 @@ const Long = require('long')
 
+const MOST_SIGNIFICANT_BIT = 0x80 // 128
+const OTHER_BITS = 0x7f // 127
+
 module.exports = class Decoder {
@@ -124,2 +127,38 @@ static int32Size() {
 
+  readSignedVarInt32() {
+    let currentByte
+    let result = 0
+    let i = 0
+
+    do {
+      currentByte = this.buffer[this.offset++]
+      result += (currentByte & OTHER_BITS) << i
+      i += 7
+    } while (currentByte >= MOST_SIGNIFICANT_BIT)
+
+    return this.decodeZigZag(result)
+  }
+
+  decodeZigZag(value) {
+    return (value >>> 1) ^ -(value & 1)
+  }
+
+  readSignedVarInt64() {
+    let currentByte
+    let result = Long.fromInt(0)
+    let i = 0
+
+    do {
+      currentByte = this.buffer[this.offset++]
+      result = result.add(Long.fromInt(currentByte & OTHER_BITS).shiftLeft(i))
+      i += 7
+    } while (currentByte >= MOST_SIGNIFICANT_BIT)
+
+    return this.decodeZigZag64(result)
+  }
+
+  decodeZigZag64(longValue) {
+    return longValue.shiftRightUnsigned(1).xor(longValue.and(Long.fromInt(1)).negate())
+  }
+
   slice(size) {
@@ -126,0 +165,0 @@ return new Decoder(this.buffer.slice(this.offset, this.offset + size))

src/protocol/encoder.js

@@ -8,2 +8,7 @@ const Long = require('long')
 
+const MOST_SIGNIFICANT_BIT = 0x80 // 128
+const OTHER_BITS = 0x7f // 127
+const UNSIGNED_INT32_MAX_NUMBER = 0xffffff80
+const UNSIGNED_INT64_MAX_NUMBER = Long.fromBytes([-1, -1, -1, -1, -1, -1, -1, -128])
+
 module.exports = class Encoder {
@@ -113,2 +118,47 @@ constructor() {
 
+  // Based on:
+  // https://github.com/addthis/stream-lib/blob/master/src/main/java/com/clearspring/analytics/util/Varint.java#L106
+  writeSignedVarInt32(value) {
+    const byteArray = []
+    let encodedValue = this.encodeZigZag(value)
+
+    while ((encodedValue & UNSIGNED_INT32_MAX_NUMBER) !== 0) {
+      byteArray.push((encodedValue & OTHER_BITS) | MOST_SIGNIFICANT_BIT)
+      encodedValue >>>= 7
+    }
+
+    byteArray.push(encodedValue & OTHER_BITS)
+    this.buffer = Buffer.concat([this.buffer, Buffer.from(byteArray)])
+    return this
+  }
+
+  encodeZigZag(value) {
+    return (value << 1) ^ (value >> 31)
+  }
+
+  writeSignedVarInt64(value) {
+    const byteArray = []
+    let longValue = this.encodeZigZag64(value)
+
+    while (longValue.and(UNSIGNED_INT64_MAX_NUMBER).notEquals(Long.fromInt(0))) {
+      byteArray.push(
+        longValue
+          .and(OTHER_BITS)
+          .or(MOST_SIGNIFICANT_BIT)
+          .toInt()
+      )
+      longValue = longValue.shiftRightUnsigned(7)
+    }
+
+    byteArray.push(longValue.toInt())
+
+    this.buffer = Buffer.concat([this.buffer, Buffer.from(byteArray)])
+    return this
+  }
+
+  encodeZigZag64(value) {
+    const longValue = Long.fromValue(value)
+    return longValue.shiftLeft(1).xor(longValue.shiftRight(63))
+  }
+
   size() {
@@ -115,0 +165,0 @@ return Buffer.byteLength(this.buffer)

No alert changes

Improved metrics

Total package byte prevSize

377505

increased by0.68%

Lines of code

5943

increased by1.28%

Dependency changes

• + Addedlong@4.0.0(transitive)

• - Removedlong@3.2.0(transitive)

• Updatedlong@^4.0.0