package com.inmobi.messaging.consumer.examples;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.HashSet;
import java.util.Set;

import com.inmobi.messaging.ClientConfig;
import com.inmobi.messaging.Message;
import com.inmobi.messaging.consumer.MessageConsumer;
import com.inmobi.messaging.consumer.MessageConsumerFactory;
import com.inmobi.messaging.publisher.AbstractMessagePublisher;
import com.inmobi.messaging.publisher.MessagePublisherFactory;

public class StreamingBenchmark {

  static final String DELIMITER = "/t";
  static final SimpleDateFormat LogDateFormat = new SimpleDateFormat(
      "yyyy:MM:dd hh:mm:ss");

  public static void main(String[] args) throws Exception {
    if (args.length != 2) {
      System.out
          .println("Usage: StreamingBenchmark <no-of-msgs> <sleepMillis-every-msg>");
      System.exit(-1);
    }
    long maxSent = Long.parseLong(args[0]);
    int sleepMillis = Integer.parseInt(args[1]);

    ClientConfig config = ClientConfig.loadFromClasspath(
        MessageConsumerFactory.MESSAGE_CLIENT_CONF_FILE);
    String topic = config.getString(MessageConsumerFactory.TOPIC_NAME_KEY);
    System.out.println("Using topic: " + topic);

    Producer producer = new Producer(topic, maxSent, sleepMillis);
    producer.start();

    Consumer consumer = new Consumer(config, maxSent);
    consumer.start();
    
    StatusLogger statusPrinter = new StatusLogger(producer, consumer);
    statusPrinter.start();

    consumer.join();
    statusPrinter.stopped = true;
    statusPrinter.join();
    if (consumer.received != consumer.seqSet.size()) {
      System.out.println("Data validation FAILED!");
    } else {
      System.out.println("Data validation SUCCESS!");
    }
    System.exit(0);
  }

  static class Producer extends Thread {
    volatile AbstractMessagePublisher publisher;
    String topic;
    long maxSent;
    int sleepMillis;

    Producer(String topic, long maxSent, int sleepMillis) throws IOException {
      this.topic = topic;
      this.maxSent = maxSent;
      this.sleepMillis = sleepMillis;
      publisher = (AbstractMessagePublisher) MessagePublisherFactory.create();
    }

    @Override
    public void run() {
      System.out.println("Producer started!");
      for (long i = 1; i <= maxSent; i++) {
        long time = System.currentTimeMillis();
        String s = i + DELIMITER + Long.toString(time);
        Message msg = new Message(ByteBuffer.wrap(s.getBytes()));
        publisher.publish(topic, msg);
        
        try {
          Thread.sleep(sleepMillis);
        } catch (InterruptedException e) {
          e.printStackTrace();
          return;
        }
      }
      try {
        Thread.sleep(10000);
      } catch (InterruptedException e) {
        e.printStackTrace();
        return;
      }
    }

  }

  static class Consumer extends Thread {
    volatile Set<Long> seqSet = new HashSet<Long>();
    final MessageConsumer consumer;
    final long maxSent;
    volatile long received = 0;
    volatile long totalLatency = 0;

    Consumer(ClientConfig config, long maxSent) throws IOException {
      this.maxSent = maxSent;
      consumer = MessageConsumerFactory.create(config);
    }

    @Override
    public void run() {
      System.out.println("Consumer started!");
      while (true) {
        Message msg = null;
        try {
          msg = consumer.next();
          received++;
          String s = new String(msg.getData().array());
          String[] ar = s.split(DELIMITER);
          long seq = Long.parseLong(ar[0]);
          seqSet.add(seq);
          long sentTime = Long.parseLong(ar[1]);
          totalLatency += System.currentTimeMillis() - sentTime;
          if (received == maxSent) {
            break;
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
          return;
        }
      }
      consumer.close();
    }

  }

  static class StatusLogger extends Thread {
    volatile boolean stopped;
    Producer producer;
    Consumer consumer;
    StatusLogger(Producer producer, Consumer consumer) {
      this.producer = producer;
      this.consumer = consumer;
    }
    @Override
    public void run() {
      while(!stopped) {
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
          e.printStackTrace();
          return;
        }
        StringBuffer sb = new StringBuffer();
        sb.append(LogDateFormat.format(System.currentTimeMillis()));
        sb.append(" Invocations:" + producer.publisher.getStats().getInvocationCount());
        sb.append(" SentSuccess:" + producer.publisher.getStats().getSuccessCount());
        sb.append(" UnhandledExceptions:" + producer.publisher.getStats().getUnhandledExceptionCount());
        sb.append(" Received:" + consumer.received);
        sb.append(" UniqueReceived:" + consumer.seqSet.size());
        if (consumer.received != 0) {
          sb.append(" MeanLatency(ms):" + (consumer.totalLatency / consumer.received));
        }
        System.out.println(sb.toString());
      }
    }
  }
}

@@ -133,16 +133,66 @@
 Format;%0A
+import java.util.Calendar;%0Aimport java.util.Date;%0A
 import j
@@ -1357,31 +1357,8 @@
 s);%0A
-    producer.start();%0A%0A
     
@@ -1405,16 +1405,80 @@
  maxSent
+,%0A        Calendar.getInstance().getTime());%0A    producer.start(
 );%0A    c
@@ -3216,17 +3216,42 @@
  maxSent
-)
+, Date startTime) %0A       
  throws 
@@ -3347,16 +3347,27 @@
 e(config
+, startTime
 );%0A    %7D

Change streaming benchmark consumer to start from 'now'