message: Move to interval
diff: @@ -146,14 +146,18 @@ async function kafkaBlizzardLatencyTest(startOffset: number, messages: number) {
  let messagesSent = 0;
  let messagesReceived = 0;
  -
  + let messagesAckd = 0;
  let totalLatency = 0;
  + let productionInterval: NodeJS.Timer;
  +
  let producer = new kafkaBlizzard.Producer({
  "client.id": clientId,
  "debug": "all",
  "dr_cb": (data) => {
  - console.log("dr_cb_" + messagesSent + ": " + Date.now());
  - messagesSent++;
  + if (commander.progress) {
  + console.log("dr_cb_" + messagesAckd + ": " + Date.now());
  + }
  + messagesAckd++;
  },
  "metadata.broker.list": kafkaEndpoint,
  }, {});
  @@ -171,34 +175,26 @@ async function kafkaBlizzardLatencyTest(startOffset: number, messages: number) {
  "metadata.broker.list": kafkaEndpoint,
  "topic": topic,
  }, (err, data) => {
  - producer.setPollInterval(10000); // How does poll time impact latency?
  + producer.setPollInterval(100); // Not required
  });
  producer.on("ready", (arg) => {
  - let sendQueue = new queue((i, cb) => {
  - setImmediate(() => {
  - console.log()
  + productionInterval = setInterval(() => {
  producer.produce(
  topic,
  0,
  new Buffer("message"),
  );
  - });
  - cb();
  - }, 1);
  -
  - for (let i = 0; i < messages; i++) {
  - sendQueue.push(i, () => {
  - console.log("Queue_CB_" + i + ": " + Date.now())
  - return undefined;
  - });
  + messagesSent++;
  + if (messagesSent === messages) {
  + clearInterval(productionInterval);
  }
  + }, 10);
  });
  consumer.on("ready", (arg) => {
  - console.log("consumerReady: " + Date.now());
  consumer.assign([{
  topic: topic,
  partition: partition,
  @@ -210,22 +206,28 @@ async function kafkaBlizzardLatencyTest(startOffset: number, messages: number) {
  consumer.on("data", (data) => {
  let now = Date.now();
  messagesReceived++;
  + if (commander.progress) {
  console.log("Sent Time " + data.timestamp);
  console.log("Received Time: " + now);
  console.log("Latency: " + (now - data.timestamp));
  + }
  totalLatency += (now - data.timestamp);
  });
  consumer.connect();
  - setInterval(() => {
  + let int = setInterval(() => {
  console.log("--------Interval--------");
  console.log("messagesSent: " + messagesSent);
  console.log("messagesReceived: " + messagesReceived);
  console.log("totalLatency: " + totalLatency);
  console.log("AvgLatency: " + (totalLatency/ messagesReceived));
  - }, 30000);
  -
  + if (messagesSent === messages && messages === messagesReceived ) {
  + consumer.disconnect()
  + producer.disconnect();
  + clearInterval(int);
  + }
  + }, 5000);
  }
  async function kafkaBlizzardProducerTest(startOffset) {
  