javascript - 索引超出范围 this.buf.utf8Slice

Question

我有一个使用 confluence.io kafka 包的带有 avro 消息的 kafka 流。这对于 java 应用程序来说工作得很好。但我现在正在尝试用 JavaScript 读取这些消息。

我一直在尝试使用 kafka-node + avsc 包使用模式将消息从缓冲区数组解码为字符串。我知道 Confluence 将前 5 个字节作为魔术字节 (0) + 4 个字节作为模式 ID。

因此，我对缓冲区进行切片以删除这些字节，并尝试将其发送到 avsc 进行解码。但我收到错误

return this.buf.utf8Slice(pos, pos + len);

RangeError: out of range index at RangeError (native) at Tap.readString (C:\git\workflowapps\workItemsApp\node_modules\avsc\lib\utils.js:452:19) at StringType._read (C:\git\workflowapps\workItemsApp\node_modules\avsc\lib\types.js:612:58)

还尝试手动解码，这会留下大量非 utf8 字符，这样我就会丢失数据。

示例代码:

  consumer.on('message', function(message) {
      var val = message.value.slice(4);
      sails.log.info('val buffer', val, val.length);
      sails.log.info('hex',val.toString('hex'));
      var type = avro.parse({"type":"record",
       "name":"StatusEvent",
        "fields":[{"name":"ApplicationUUID","type":"string"},
        {"name":"StatusUUID","type":"string"},
        {"name":"Name","type":"string"},
        {"name":"ChangedBy","type":"string"},
        {"name":"ChangedByUUID","type":"string"},
        {"name":"ChangedAt","type":"long"}]
      });

      var decodedValue = type.fromBuffer(val);
      sails.log.info('Decoded', decodedValue);
    });

Answer 1

您的 slice(4) 应该是 slice(5) (否则您只会跳过 5 个 header 字节中的 4 个)。您也许还可以找到有用的信息 here .