package cn.hashdata.bireme;

import java.util.Date;
import java.util.HashMap;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;

import cn.hashdata.bireme.pipeline.PipeLine;

/**
 * {@code PipeLineStat} contains the statistics about the {@code PipeLine}.
 *
 * @author yuze
 *
 */
public class PipeLineStat {
  private String pipeLineName;
  public Long newestCompleted = 0L;
  public Long delay = 0L;

  private PipeLine pipeLine;
  private MetricRegistry register;

  public Timer avgDelay;
  public Meter recordCount;
  public Gauge<String> completed;
  public Gauge<Long> transformQueueSize;

  public HashMap<String, Gauge<Long>> cacheSize;
  public HashMap<String, Timer> copyForDelete;
  public HashMap<String, Timer> delete;
  public HashMap<String, Timer> insert;

  public PipeLineStat(PipeLine pipeLine) {
    this.pipeLine = pipeLine;
    this.pipeLineName = pipeLine.myName;
    this.register = pipeLine.cxt.register;

    cacheSize = new HashMap<String, Gauge<Long>>();
    copyForDelete = new HashMap<String, Timer>();
    delete = new HashMap<String, Timer>();
    insert = new HashMap<String, Timer>();

    avgDelay = register.timer(MetricRegistry.name(pipeLineName, "AvgDelay"));
    recordCount = register.meter(MetricRegistry.name(pipeLineName, "RecordCount"));

    completed = new Gauge<String>() {

      @Override
      public String getValue() {
        return new Date(newestCompleted).toString();
      }
    };

    transformQueueSize = new Gauge<Long>() {

      @Override
      public Long getValue() {
        return (long) pipeLine.transResult.size();
      }

    };

    register.register(MetricRegistry.name(pipeLineName, "Completed"), completed);
    register.register(MetricRegistry.name(pipeLineName, "TransformQueueSize"), transformQueueSize);
  }

  /**
   * Add {@code Gauge} for a {@link RowCache} to get the size.
   *
   * @param table the name of the cache
   * @param cache the {@code RowCache} itselff
   */
  public void addGaugeForCache(String table, RowCache cache) {
    Gauge<Long> gauge = new Gauge<Long>() {

      @Override
      public Long getValue() {
        return (long) cache.rows.size();
      }

    };

    cacheSize.put(table, gauge);
    register.register(MetricRegistry.name(pipeLineName, table, "Cache"), gauge);
  }

  /**
   * Add {@code Timer}s for a {@link ChangeLoader}. Each {@code ChangeLoader} has three Timer.
   * <ul>
   * <li>Record the time of copy followed by delete</li>
   * <li>Record the time of delete</li>
   * <li>Record the time of insert in copy way</li>
   * </ul>
   *
   * @param table the {@code ChangeLoader}'s table
   * @return the registered Timer
   */
  public Timer[] addTimerForLoader(String table) {
    Timer[] timers = new Timer[3];
    timers[0] = register.timer(MetricRegistry.name(pipeLineName, table, "Loader", "CopyForDelete"));
    timers[1] = register.timer(MetricRegistry.name(pipeLineName, table, "Loader", "Delete"));
    timers[2] = register.timer(MetricRegistry.name(pipeLineName, table, "Loader", "Insert"));

    copyForDelete.put(table, timers[0]);
    delete.put(table, timers[1]);
    insert.put(table, timers[2]);
    return timers;
  }
}

@@ -402,16 +402,76 @@
 eStat %7B%0A
+  private static final Long SCECOND_TO_MILLISECOND = 1000L;%0A
   privat
@@ -493,16 +493,16 @@
 neName;%0A
-
   public
@@ -708,16 +708,48 @@
 pleted;%0A
+  public Gauge%3CString%3E syncGap;%0A
   public
@@ -1600,20 +1600,280 @@
 eted
-).toString()
+ * SCECOND_TO_MILLISECOND).toString();%0A      %7D%0A    %7D;%0A%0A    syncGap = new Gauge%3CString%3E() %7B%0A%0A      @Override%0A      public String getValue() %7B%0A        Long now = new Date().getTime() / SCECOND_TO_MILLISECOND;%0A        return String.valueOf(now - newestCompleted) + %22seconds.%22
 ;%0A  
@@ -2043,24 +2043,24 @@
  %7D%0A%0A    %7D;%0A%0A
-
     register
@@ -2129,16 +2129,94 @@
 leted);%0A
+    register.register(MetricRegistry.name(pipeLineName, %22SyncGap%22), syncGap);%0A
     regi

Add sync gap metrics