<commit_before>'use strict';

var Binary = require('binary');
var Buffermaker = require('buffermaker');
var _ = require('lodash');
var crc32 = require('buffer-crc32');
var protocol = require('./protocol_struct');
var getCodec = require('../codec');
var REQUEST_TYPE = protocol.REQUEST_TYPE;
var ERROR_CODE = protocol.ERROR_CODE;
var GROUP_ERROR = protocol.GROUP_ERROR;
var PartitionMetadata = protocol.PartitionMetadata;

var API_VERSION = 0;
var REPLICA_ID = -1;
var GROUPS_PROTOCOL_TYPE = 'consumer';

function groupByTopic (payloads) {
  return payloads.reduce(function (out, p) {
    out[p.topic] = out[p.topic] || {};
    out[p.topic][p.partition] = p;
    return out;
  }, {});
}

function encodeRequestWithLength (request) {
  return new Buffermaker()
    .Int32BE(request.length)
    .string(request)
    .make();
}

function encodeRequestHeader (clientId, correlationId, apiKey, apiVersion) {
  return new Buffermaker()
    .Int16BE(apiKey)
    .Int16BE(apiVersion || API_VERSION)
    .Int32BE(correlationId)
    .Int16BE(clientId.length)
    .string(clientId);
}

function encodeFetchRequest (maxWaitMs, minBytes) {
  return function encodeFetchRequest (clientId, correlationId, payloads) {
    return _encodeFetchRequest(clientId, correlationId, payloads, maxWaitMs, minBytes);
  };
}

function decodeTopics (decodePartitions) {
  return function (end, vars) {
    if (--vars.topicNum === 0) end();
    this.word16bs('topic')
      .tap(function (vars) {
        this.buffer('topic', vars.topic);
        vars.topic = vars.topic.toString();
      })
      .word32bs('partitionNum')
      .loop(decodePartitions);
  };
}

function _encodeFetchRequest (clientId, correlationId, payloads, maxWaitMs, minBytes) {
  payloads = groupByTopic(payloads);
  var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.fetch);
  var topics = Object.keys(payloads);

  request.Int32BE(REPLICA_ID)
    .Int32BE(maxWaitMs)
    .Int32BE(minBytes)
    .Int32BE(topics.length);

  topics.forEach(function (topic) {
    request.Int16BE(topic.length)
      .string(topic);

    var partitions = _.toPairs(payloads[topic]).map(function (pairs) { return pairs[1]; });
    request.Int32BE(partitions.length);
    partitions.forEach(function (p) {
      request.Int32BE(p.partition)
        .Int64BE(p.offset)
        .Int32BE(p.maxBytes);
    });
  });

  return encodeRequestWithLength(request.make());
}

function decodeFetchResponse (cb, maxTickMessages) {
  return function (resp) {
    return _decodeFetchResponse(resp, cb, maxTickMessages);
  };
}

function createGroupError (errorCode) {
  if (errorCode == null || errorCode === 0) {
    return null;
  }

  var error = ERROR_CODE[errorCode];
  if (error in GROUP_ERROR) {
    error = new GROUP_ERROR[error]('Kafka Error Code: ' + errorCode);
  } else {
    error = new Error(error);
  }
  error.errorCode = errorCode;
  return error;
}

function _decodeFetchResponse (resp, cb, maxTickMessages) {
  var topics = {};
  Binary.parse(resp)
    .word32bs('size')
    .word32bs('correlationId')
    .word32bs('topicNum')
    .loop(decodeTopics(decodePartitions));

  function decodePartitions (end, vars) {
    if (--vars.partitionNum === 0) end();
    topics[vars.topic] = topics[vars.topic] || {};
    this.word32bs('partition')
      .word16bs('errorCode')
      .word64bs('highWaterOffset')
      .word32bs('messageSetSize')
      .tap(function (vars) {
        this.buffer('messageSet', vars.messageSetSize);
        if (vars.errorCode !== 0) {
          return cb({ topic: vars.topic, partition: vars.partition, message: ERROR_CODE[vars.errorCode] });
        }
        var messageSet = decodeMessageSet(vars.topic, vars.partition, vars.messageSet, cb, maxTickMessages);
        if (messageSet.length) {
          var offset = messageSet[messageSet.length - 1];
          topics[vars.topic][vars.partition] = offset;
        }
      });
  }
  cb && cb(null, 'done', topics);
}

function decodeMessageSet (topic, partition, messageSet, cb, maxTickMessages) {
  var set = [];
  var messageCount = 0;
  while (messageSet.length > 0) {
    var cur = 8 + 4 + 4 + 1 + 1 + 4 + 4;
    Binary.parse(messageSet)
      .word64bs('offset')
      .word32bs('messageSize')
      .tap(function (vars) {
        if (vars.messageSize > (messageSet.length - 12)) {
          vars.partial = true;
        }
      })
      .word32bs('crc')
      .word8bs('magicByte')
      .word8bs('attributes')
      .word32bs('key')
      .tap(function (vars) {
        if (vars.key === -1) return;
        cur += vars.key;
        this.buffer('key', vars.key);
      })
      .word32bs('value')
      .tap(function (vars) {
        if (vars.value !== -1) {
          cur += vars.value;
          this.buffer('value', vars.value);
        } else {
          vars.value = null;
        }

        if (!vars.partial && vars.offset !== null) {
          messageCount++;
          set.push(vars.offset);
          if (!cb) return;
          var codec = getCodec(vars.attributes);
          if (!codec) {
            return cb(null, 'message', {
              topic: topic,
              value: vars.value,
              offset: vars.offset,
              partition: partition,
              key: vars.key
            });
          }
          codec.decode(vars.value, function (error, inlineMessageSet) {
            if (error) return; // Not sure where to report this
            decodeMessageSet(topic, partition, inlineMessageSet, cb, maxTickMessages);
          });
        }
      });
    // Defensive code around potential denial of service<commit_msg>Passing highWaterOffset both to messages and on Done (#610)<commit_after>'use strict';

var Binary = require('binary');
var Buffermaker = require('buffermaker');
var _ = require('lodash');
var crc32 = require('buffer-crc32');
var protocol = require('./protocol_struct');
var getCodec = require('../codec');
var REQUEST_TYPE = protocol.REQUEST_TYPE;
var ERROR_CODE = protocol.ERROR_CODE;
var GROUP_ERROR = protocol.GROUP_ERROR;
var PartitionMetadata = protocol.PartitionMetadata;

var API_VERSION = 0;
var REPLICA_ID = -1;
var GROUPS_PROTOCOL_TYPE = 'consumer';

function groupByTopic (payloads) {
  return payloads.reduce(function (out, p) {
    out[p.topic] = out[p.topic] || {};
    out[p.topic][p.partition] = p;
    return out;
  }, {});
}

function encodeRequestWithLength (request) {
  return new Buffermaker()
    .Int32BE(request.length)
    .string(request)
    .make();
}

function encodeRequestHeader (clientId, correlationId, apiKey, apiVersion) {
  return new Buffermaker()
    .Int16BE(apiKey)
    .Int16BE(apiVersion || API_VERSION)
    .Int32BE(correlationId)
    .Int16BE(clientId.length)
    .string(clientId);
}

function encodeFetchRequest (maxWaitMs, minBytes) {
  return function encodeFetchRequest (clientId, correlationId, payloads) {
    return _encodeFetchRequest(clientId, correlationId, payloads, maxWaitMs, minBytes);
  };
}

function decodeTopics (decodePartitions) {
  return function (end, vars) {
    if (--vars.topicNum === 0) end();
    this.word16bs('topic')
      .tap(function (vars) {
        this.buffer('topic', vars.topic);
        vars.topic = vars.topic.toString();
      })
      .word32bs('partitionNum')
      .loop(decodePartitions);
  };
}

function _encodeFetchRequest (clientId, correlationId, payloads, maxWaitMs, minBytes) {
  payloads = groupByTopic(payloads);
  var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.fetch);
  var topics = Object.keys(payloads);

  request.Int32BE(REPLICA_ID)
    .Int32BE(maxWaitMs)
    .Int32BE(minBytes)
    .Int32BE(topics.length);

  topics.forEach(function (topic) {
    request.Int16BE(topic.length)
      .string(topic);

    var partitions = _.toPairs(payloads[topic]).map(function (pairs) { return pairs[1]; });
    request.Int32BE(partitions.length);
    partitions.forEach(function (p) {
      request.Int32BE(p.partition)
        .Int64BE(p.offset)
        .Int32BE(p.maxBytes);
    });
  });

  return encodeRequestWithLength(request.make());
}

function decodeFetchResponse (cb, maxTickMessages) {
  return function (resp) {
    return _decodeFetchResponse(resp, cb, maxTickMessages);
  };
}

function createGroupError (errorCode) {
  if (errorCode == null || errorCode === 0) {
    return null;
  }

  var error = ERROR_CODE[errorCode];
  if (error in GROUP_ERROR) {
    error = new GROUP_ERROR[error]('Kafka Error Code: ' + errorCode);
  } else {
    error = new Error(error);
  }
  error.errorCode = errorCode;
  return error;
}

function _decodeFetchResponse (resp, cb, maxTickMessages) {
  var topics = {};
  Binary.parse(resp)
    .word32bs('size')
    .word32bs('correlationId')
    .word32bs('topicNum')
    .loop(decodeTopics(decodePartitions));

  function decodePartitions (end, vars) {
    if (--vars.partitionNum === 0) end();
    topics[vars.topic] = topics[vars.topic] || {};
    this.word32bs('partition')
      .word16bs('errorCode')
      .word64bs('highWaterOffset')
      .word32bs('messageSetSize')
      .tap(function (vars) {
        this.buffer('messageSet', vars.messageSetSize);
        if (vars.errorCode !== 0) {
          return cb({ topic: vars.topic, partition: vars.partition, message: ERROR_CODE[vars.errorCode] });
        }
        var messageSet = decodeMessageSet(vars.topic, vars.partition, vars.messageSet, cb, maxTickMessages, vars.highWaterOffset);
        if (messageSet.length) {
          var offset = messageSet[messageSet.length - 1];
          topics[vars.topic][vars.partition] = offset;
          topics[vars.topic].highWaterOffset = vars.highWaterOffset;
        }
      });
  }
  cb && cb(null, 'done', topics);
}

function decodeMessageSet (topic, partition, messageSet, cb, maxTickMessages, highWaterOffset) {
  var set = [];
  var messageCount = 0;
  while (messageSet.length > 0) {
    var cur = 8 + 4 + 4 + 1 + 1 + 4 + 4;
    Binary.parse(messageSet)
      .word64bs('offset')
      .word32bs('messageSize')
      .tap(function (vars) {
        if (vars.messageSize > (messageSet.length - 12)) {
          vars.partial = true;
        }
      })
      .word32bs('crc')
      .word8bs('magicByte')
      .word8bs('attributes')
      .word32bs('key')
      .tap(function (vars) {
        if (vars.key === -1) return;
        cur += vars.key;
        this.buffer('key', vars.key);
      })
      .word32bs('value')
      .tap(function (vars) {
        if (vars.value !== -1) {
          cur += vars.value;
          this.buffer('value', vars.value);
        } else {
          vars.value = null;
        }

        if (!vars.partial && vars.offset !== null) {
          messageCount++;
          set.push(vars.offset);
          if (!cb) return;
          var codec = getCodec(vars.attributes);
          if (!codec) {
            return cb(null, 'message', {
              topic: topic,
              value: vars.value,
              offset: vars.offset,
              partition: partition,
              highWaterOffset: highWaterOffset,
              key: vars.key
            });
          }
          codec.decode(vars.value, function (error, inlineMessageSet) {
            if (error) return; // Not sure where to report this
            decodeMessageSet(topic, partition, inlineMessageSet, cb, maxTickMessages);
          });
        }
      });
    // Defensive code around potential denial of service
{'fraction_non_alphanumeric': 0.13753959873284055, 'fraction_numerical': 0.01487152411122844, 'mean_word_length': 2.8814890710382515, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1746308', 'n_tokens_mistral': 3674, 'n_tokens_neox': 3517, 'n_words': 749}