
package org.usergrid.benchmark.commands.queue;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.commons.cli.CommandLine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.usergrid.event.Queue;
import org.usergrid.tools.event.EventListener;
import org.usergrid.tools.event.TestEvent;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Timer;

/**
 * A benchmark util used to read all data from a topic and dump it to a file for
 * analysis
 * 
 * @author tnine
 * 
 */
public class QueueReader extends QueueBase {

  protected static final Logger readLogger = LoggerFactory.getLogger("read");

  protected static final Timer readsTimer = Metrics.newTimer(QueueReader.class, "reads", TimeUnit.MILLISECONDS,
      TimeUnit.SECONDS);

  
  @Autowired
  private HazelcastInstance instance;
  
  /*
   * (non-Javadoc)
   * 
   * @see org.usergrid.tools.TopicBase#doWork(org.usergrid.event.Topic,
   * java.lang.String, int)
   */
  @Override
  protected void doWork(CommandLine line, Queue<TestEvent> queue, String hostName, int workers, int count)
      throws Exception {

    CountDownLatch latch = new CountDownLatch(count * workers);
    
    queue.observe(new EventListener(latch, readsTimer, readLogger, queue));
    
    latch.await();

    writeTimerData(readsTimer);

  }

}
