code: package resa.metrics;
  
  import backtype.storm.Config;
  import backtype.storm.metric.api.MultiCountMetric;
  import backtype.storm.task.IOutputCollector;
  import backtype.storm.task.OutputCollector;
  import backtype.storm.task.TopologyContext;
  import backtype.storm.topology.IRichBolt;
  import backtype.storm.tuple.Tuple;
  import backtype.storm.utils.Utils;
  import resa.topology.DelegatedBolt;
  import resa.util.ConfigUtil;
  import resa.util.ResaConfig;
  import resa.util.Sampler;
  
  import java.util.Collection;
  import java.util.List;
  import java.util.Map;
  
  /**
   * Created by ding on 14-1-27.
   */
  public class MeasurableBolt extends DelegatedBolt {
  
      private class MeasurableOutputCollector extends OutputCollector {
  
          private boolean sample = false;
  
          MeasurableOutputCollector(IOutputCollector delegate) {
              super(delegate);
          }
  
          public void setEmitSample(boolean sample) {
              this.sample = sample;
          }
  
          @Override
          public List<Integer> emit(String streamId, Collection<Tuple> anchors, List<Object> tuple) {
              if (sample) {
                  emitMetric.scope(streamId).incr();
              }
              return super.emit(streamId, anchors, tuple);
          }
  
          @Override
          public void emitDirect(int taskId, String streamId, Collection<Tuple> anchors, List<Object> tuple) {
              if (sample) {
                  emitMetric.scope(streamId).incr();
              }
              super.emitDirect(taskId, streamId, anchors, tuple);
          }
      }
  
      private transient CMVMetric executeMetric;
      private Sampler sampler;
      private transient MultiCountMetric emitMetric;
      private transient MeasurableOutputCollector measurableCollector;
      private long lastMetricsSent;
  
      public MeasurableBolt(IRichBolt delegate) {
          super(delegate);
      }
  
      @Override
      public void prepare(Map conf, TopologyContext context, OutputCollector outputCollector) {
          int interval = Utils.getInt(conf.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS));
          executeMetric = context.registerMetric(MetricNames.TASK_EXECUTE, new CMVMetric(), interval);
          emitMetric = context.registerMetric(MetricNames.EMIT_COUNT, new MultiCountMetric(), interval);
          lastMetricsSent = System.currentTimeMillis();
          context.registerMetric(MetricNames.DURATION, this::getMetricsDuration, interval);
          sampler = new Sampler(ConfigUtil.getDouble(conf, ResaConfig.COMP_SAMPLE_RATE, 0.05));
          measurableCollector = new MeasurableOutputCollector(outputCollector);
          super.prepare(conf, context, measurableCollector);
      }
  
      private long getMetricsDuration() {
          long now = System.currentTimeMillis();
          long duration = now - lastMetricsSent;
          lastMetricsSent = now;
          return duration;
      }
  
      @Override
      public void execute(Tuple tuple) {
          long elapse;
          if (sampler.shoudSample()) {
              // enable emit sample
              measurableCollector.setEmitSample(true);
              long arrivalTime = System.nanoTime();
              super.execute(tuple);
              elapse = System.nanoTime() - arrivalTime;
          } else {
              elapse = -1;
              // disable emit sample
              measurableCollector.setEmitSample(false);
              super.execute(tuple);
          }
          // avoid numerical overflow
          if (elapse > 0) {
              String id = tuple.getSourceComponent() + ":" + tuple.getSourceStreamId();
              executeMetric.addMetric(id, elapse / 1000000.0);
  //            System.out.println("from " + tuple.getSourceComponent() + ":" + tuple.getSourceStreamId() +
  //                    " cost->" + (elapse / 1000000.0));
          }
      }
  
  }
  