/**
 * Returns the value of a metric, or {@code null} if not found.
 */
@Nullable
public Double getMetric(Measure measure) {
  for (Map.Entry<Measure, Number> m : metrics.entrySet()) {
    if (m.getKey().equals(measure)) {
      Number value = m.getValue();
      if (value instanceof Double) {
        return (Double) value;
      } else if (value instanceof Long) {
        return (double) (Long) value;
      }
      throw new AssertionError("Unexpected measure value type: " + value.getClass().getName());
    }
  }
  return null;
}
 

@javax.annotation.Nullable
static MetricDescriptor viewToMetricDescriptor(View view) {
  if (view.getWindow() instanceof View.AggregationWindow.Interval) {
    // Only creates Metric for cumulative stats.
    return null;
  }
  List<LabelKey> labelKeys = new ArrayList<LabelKey>();
  for (TagKey tagKey : view.getColumns()) {
    // TODO: add description
    labelKeys.add(LabelKey.create(tagKey.getName(), ""));
  }
  Measure measure = view.getMeasure();
  Aggregation aggregation = view.getAggregation();
  return MetricDescriptor.create(
      view.getName().asString(),
      view.getDescription(),
      getUnit(measure, aggregation),
      getType(measure, aggregation),
      labelKeys);
}
 

@VisibleForTesting
static Type getType(Measure measure, Aggregation aggregation) {
  return aggregation.match(
      Functions.returnConstant(
          measure.match(
              TYPE_CUMULATIVE_DOUBLE_FUNCTION, // Sum Double
              TYPE_CUMULATIVE_INT64_FUNCTION, // Sum Int64
              TYPE_UNRECOGNIZED_FUNCTION)),
      TYPE_CUMULATIVE_INT64_FUNCTION, // Count
      TYPE_CUMULATIVE_DISTRIBUTION_FUNCTION, // Distribution
      Functions.returnConstant(
          measure.match(
              TYPE_GAUGE_DOUBLE_FUNCTION, // LastValue Double
              TYPE_GAUGE_INT64_FUNCTION, // LastValue Long
              TYPE_UNRECOGNIZED_FUNCTION)),
      AGGREGATION_TYPE_DEFAULT_FUNCTION);
}
 

/** Enable stats collection for the given {@link View}. */
synchronized void registerView(View view, Clock clock) {
  exportedViews = null;
  View existing = registeredViews.get(view.getName());
  if (existing != null) {
    if (existing.equals(view)) {
      // Ignore views that are already registered.
      return;
    } else {
      throw new IllegalArgumentException(
          "A different view with the same name is already registered: " + existing);
    }
  }
  Measure measure = view.getMeasure();
  Measure registeredMeasure = registeredMeasures.get(measure.getName());
  if (registeredMeasure != null && !registeredMeasure.equals(measure)) {
    throw new IllegalArgumentException(
        "A different measure with the same name is already registered: " + registeredMeasure);
  }
  registeredViews.put(view.getName(), view);
  if (registeredMeasure == null) {
    registeredMeasures.put(measure.getName(), measure);
  }
  Timestamp now = clock.now();
  mutableMap.put(view.getMeasure().getName(), MutableViewData.create(view, now));
}
 

synchronized void record(TagContext tags, MeasureMapInternal stats, Timestamp timestamp) {
  Iterator<Measurement> iterator = stats.iterator();
  Map<String, AttachmentValue> attachments = stats.getAttachments();
  while (iterator.hasNext()) {
    Measurement measurement = iterator.next();
    Measure measure = measurement.getMeasure();
    if (!measure.equals(registeredMeasures.get(measure.getName()))) {
      // unregistered measures will be ignored.
      continue;
    }
    Collection<MutableViewData> viewDataCollection = mutableMap.get(measure.getName());
    for (MutableViewData viewData : viewDataCollection) {
      viewData.record(
          tags, RecordUtils.getDoubleValueFromMeasurement(measurement), timestamp, attachments);
    }
  }
}
 

private void testRecordCumulative(Measure measure, Aggregation aggregation, double... values) {
  View view = createCumulativeView(VIEW_NAME, measure, aggregation, Arrays.asList(KEY));
  clock.setTime(Timestamp.create(1, 2));
  viewManager.registerView(view);
  TagContext tags = tagger.emptyBuilder().put(KEY, VALUE).build();
  for (double val : values) {
    putToMeasureMap(statsRecorder.newMeasureMap(), measure, val).record(tags);
  }
  clock.setTime(Timestamp.create(3, 4));
  ViewData viewData = viewManager.getView(VIEW_NAME);
  assertThat(viewData.getView()).isEqualTo(view);
  assertThat(viewData.getWindowData())
      .isEqualTo(CumulativeData.create(Timestamp.create(1, 2), Timestamp.create(3, 4)));
  StatsTestUtil.assertAggregationMapEquals(
      viewData.getAggregationMap(),
      ImmutableMap.of(
          Arrays.asList(VALUE),
          StatsTestUtil.createAggregationData(aggregation, measure, values)),
      EPSILON);
}
 

/**
 * Returns the value of a metric, or {@code null} if not found.
 */
@Nullable
public Double getMetric(Measure measure) {
  for (Map.Entry<Measure, Number> m : metrics.entrySet()) {
    if (m.getKey().equals(measure)) {
      Number value = m.getValue();
      if (value instanceof Double) {
        return (Double) value;
      } else if (value instanceof Long) {
        return (double) (Long) value;
      }
      throw new AssertionError("Unexpected measure value type: " + value.getClass().getName());
    }
  }
  return null;
}
 

/**
 * Returns the value of a metric converted to long, or throw if not found.
 */
public long getMetricAsLongOrFail(Measure measure) {
  Double doubleValue = getMetric(measure);
  checkNotNull(doubleValue, "Measure not found: %s", measure.getName());
  long longValue = (long) (Math.abs(doubleValue) + 0.0001);
  if (doubleValue < 0) {
    longValue = -longValue;
  }
  return longValue;
}
 

private void setupView(Measure measure, TagKey... keys) {
  // Register the view. It is imperative that this step exists,
  // otherwise recorded metrics will be dropped and never exported.
  View view = View.create(
      View.Name.create(NAME_BASE + measure.getName()),
      measure.getDescription(),
      measure,
      Aggregation.Count.create(),
      ImmutableList.copyOf(keys));

  viewManager.registerView(view);
}
 

@SuppressWarnings("try")
public static void recordStat(Measure.MeasureLong ml, Long n) {
  TagContext tctx = tagger.emptyBuilder().build();
  try (Scope ss = tagger.withTagContext(tctx)) {
    statsRecorder.newMeasureMap().put(ml, n).record();
  }
}
 

@SuppressWarnings("try")
public static void recordTaggedStat(TagKey key, String value, Measure.MeasureLong ml, Long n) {
  TagContext tctx = tagger.emptyBuilder().putLocal(key, TagValue.create(value)).build();
  try (Scope ss = tagger.withTagContext(tctx)) {
    statsRecorder.newMeasureMap().put(ml, n).record();
  }
}
 

@SuppressWarnings("try")
public static void recordTaggedStat(
    TagKey key, String value, Measure.MeasureDouble md, Double d) {
  TagContext tctx = tagger.emptyBuilder().putLocal(key, TagValue.create(value)).build();
  try (Scope ss = tagger.withTagContext(tctx)) {
    statsRecorder.newMeasureMap().put(md, d).record();
  }
}
 

@SuppressWarnings("try")
public static void recordTaggedStat(
    TagKey[] keys, String[] values, Measure.MeasureDouble md, Double d) {
  TagContextBuilder builder = tagger.emptyBuilder();
  for (int i = 0; i < keys.length; i++) {
    builder.putLocal(keys[i], TagValue.create(values[i]));
  }
  TagContext tctx = builder.build();
  try (Scope ss = tagger.withTagContext(tctx)) {
    statsRecorder.newMeasureMap().put(md, d).record();
  }
}
 

@SuppressWarnings("try")
public static void recordTaggedStat(
    TagKey[] keys, String[] values, Measure.MeasureLong md, Long n) {
  TagContextBuilder builder = tagger.emptyBuilder();
  for (int i = 0; i < keys.length; i++) {
    builder.putLocal(keys[i], TagValue.create(values[i]));
  }
  TagContext tctx = builder.build();

  try (Scope ss = tagger.withTagContext(tctx)) {
    statsRecorder.newMeasureMap().put(md, n).record();
  }
}
 

private static View[] createViews(
    int size, Measure[] measures, Aggregation aggregation, TagKey... keys) {
  View[] views = new View[size];
  for (int i = 0; i < size; i++) {
    views[i] = createView(measures[i].getName(), measures[i], aggregation, keys);
  }
  return views;
}
 

private static Measure.MeasureDouble[] createMeasureDoubles(int size, String name) {
  Measure.MeasureDouble[] measures = new Measure.MeasureDouble[size];
  for (int i = 0; i < size; i++) {
    measures[i] = Measure.MeasureDouble.create(name + "_MD" + i, "", "ns");
  }
  return measures;
}
 

private static Measure.MeasureLong[] createMeasureLongs(int size, String name) {
  Measure.MeasureLong[] measures = new Measure.MeasureLong[size];
  for (int i = 0; i < size; i++) {
    measures[i] = Measure.MeasureLong.create(name + "_ML" + i, "", "ns");
  }
  return measures;
}
 

private static MeasureMap record(Data data, Measure.MeasureLong measure, int value) {
  MeasureMap map = data.recorder.newMeasureMap();
  map.put(measure, value);
  for (TagContext tags : data.contexts) {
    map.record(tags);
  }
  return map;
}
 

private static MeasureMap record(Data data, Measure.MeasureDouble measure, double value) {
  MeasureMap map = data.recorder.newMeasureMap();
  map.put(measure, value);
  for (TagContext tags : data.contexts) {
    map.record(tags);
  }
  return map;
}
 

static <T> Map<T, AggregationData> createAggregationMap(
    Map<T, MutableAggregation> tagValueAggregationMap, Measure measure) {
  Map<T, AggregationData> map = Maps.newHashMap();
  for (Entry<T, MutableAggregation> entry : tagValueAggregationMap.entrySet()) {
    map.put(entry.getKey(), entry.getValue().toAggregationData());
  }
  return map;
}
 

IntervalBucket(Timestamp start, Duration duration, Aggregation aggregation, Measure measure) {
  this.start = checkNotNull(start, "Start");
  this.duration = checkNotNull(duration, "Duration");
  checkArgument(duration.compareTo(ZERO) > 0, "Duration must be positive");
  this.aggregation = checkNotNull(aggregation, "Aggregation");
  this.measure = checkNotNull(measure, "measure");
}
 

private Map<List</*@Nullable*/ TagValue>, AggregationData> combineBucketsAndGetAggregationMap(
    Timestamp now) {
  // Need to maintain the order of inserted MutableAggregations (inserted based on time order).
  Multimap<List</*@Nullable*/ TagValue>, MutableAggregation> multimap =
      LinkedHashMultimap.create();

  ArrayDeque<IntervalBucket> shallowCopy = new ArrayDeque<IntervalBucket>(buckets);

  Aggregation aggregation = super.view.getAggregation();
  Measure measure = super.view.getMeasure();
  putBucketsIntoMultiMap(shallowCopy, multimap, aggregation, measure, now);
  Map<List</*@Nullable*/ TagValue>, MutableAggregation> singleMap =
      aggregateOnEachTagValueList(multimap, aggregation, measure);
  return createAggregationMap(singleMap, super.getView().getMeasure());
}
 

private static void putBucketsIntoMultiMap(
    ArrayDeque<IntervalBucket> buckets,
    Multimap<List</*@Nullable*/ TagValue>, MutableAggregation> multimap,
    Aggregation aggregation,
    Measure measure,
    Timestamp now) {
  // Put fractional stats of the head (oldest) bucket.
  IntervalBucket head = CheckerFrameworkUtils.castNonNull(buckets.peekFirst());
  IntervalBucket tail = CheckerFrameworkUtils.castNonNull(buckets.peekLast());
  double fractionTail = tail.getFraction(now);
  // TODO(songya): decide what to do when time goes backwards
  checkArgument(
      0.0 <= fractionTail && fractionTail <= 1.0,
      "Fraction " + fractionTail + " should be within [0.0, 1.0].");
  double fractionHead = 1.0 - fractionTail;
  putFractionalMutableAggregationsToMultiMap(
      head.getTagValueAggregationMap(), multimap, aggregation, measure, fractionHead);

  // Put whole data of other buckets.
  boolean shouldSkipFirst = true;
  for (IntervalBucket bucket : buckets) {
    if (shouldSkipFirst) {
      shouldSkipFirst = false;
      continue; // skip the first bucket
    }
    for (Entry<List</*@Nullable*/ TagValue>, MutableAggregation> entry :
        bucket.getTagValueAggregationMap().entrySet()) {
      multimap.put(entry.getKey(), entry.getValue());
    }
  }
}
 

private static <T> void putFractionalMutableAggregationsToMultiMap(
    Map<T, MutableAggregation> mutableAggrMap,
    Multimap<T, MutableAggregation> multimap,
    Aggregation aggregation,
    Measure measure,
    double fraction) {
  for (Entry<T, MutableAggregation> entry : mutableAggrMap.entrySet()) {
    // Initially empty MutableAggregations.
    MutableAggregation fractionalMutableAgg = createMutableAggregation(aggregation, measure);
    fractionalMutableAgg.combine(entry.getValue(), fraction);
    multimap.put(entry.getKey(), fractionalMutableAgg);
  }
}
 

private static <T> Map<T, MutableAggregation> aggregateOnEachTagValueList(
    Multimap<T, MutableAggregation> multimap, Aggregation aggregation, Measure measure) {
  Map<T, MutableAggregation> map = Maps.newHashMap();
  for (T tagValues : multimap.keySet()) {
    // Initially empty MutableAggregations.
    MutableAggregation combinedAggregation = createMutableAggregation(aggregation, measure);
    for (MutableAggregation mutableAggregation : multimap.get(tagValues)) {
      combinedAggregation.combine(mutableAggregation, 1.0);
    }
    map.put(tagValues, combinedAggregation);
  }
  return map;
}
 

/**
 * Creates an {@link AggregationData} by adding the given sequence of values, based on the
 * definition of the given {@link Aggregation}.
 *
 * @param aggregation the {@code Aggregation} to apply the values to.
 * @param values the values to add to the {@code MutableAggregation}s.
 * @return an {@code AggregationData}.
 */
static AggregationData createAggregationData(
    Aggregation aggregation, Measure measure, double... values) {
  MutableAggregation mutableAggregation =
      RecordUtils.createMutableAggregation(aggregation, measure);
  for (double value : values) {
    mutableAggregation.add(value, Collections.<String, AttachmentValue>emptyMap(), EMPTY);
  }
  return mutableAggregation.toAggregationData();
}
 

private void testRecord_MeasureNotMatch(Measure measure1, Measure measure2, double value) {
  viewManager.registerView(createCumulativeView(VIEW_NAME, measure1, MEAN, Arrays.asList(KEY)));
  TagContext tags = tagger.emptyBuilder().put(KEY, VALUE).build();
  putToMeasureMap(statsRecorder.newMeasureMap(), measure2, value).record(tags);
  ViewData view = viewManager.getView(VIEW_NAME);
  assertThat(view.getAggregationMap()).isEmpty();
}
 

private void testMultipleViews_DifferentMeasures(
    Measure measure1, Measure measure2, double value1, double value2) {
  final View view1 = createCumulativeView(VIEW_NAME, measure1, DISTRIBUTION, Arrays.asList(KEY));
  final View view2 =
      createCumulativeView(VIEW_NAME_2, measure2, DISTRIBUTION, Arrays.asList(KEY));
  clock.setTime(Timestamp.create(1, 0));
  viewManager.registerView(view1);
  clock.setTime(Timestamp.create(2, 0));
  viewManager.registerView(view2);
  TagContext tags = tagger.emptyBuilder().put(KEY, VALUE).build();
  MeasureMap measureMap = statsRecorder.newMeasureMap();
  putToMeasureMap(measureMap, measure1, value1);
  putToMeasureMap(measureMap, measure2, value2);
  measureMap.record(tags);
  clock.setTime(Timestamp.create(3, 0));
  ViewData viewData1 = viewManager.getView(VIEW_NAME);
  clock.setTime(Timestamp.create(4, 0));
  ViewData viewData2 = viewManager.getView(VIEW_NAME_2);
  assertThat(viewData1.getWindowData())
      .isEqualTo(CumulativeData.create(Timestamp.create(1, 0), Timestamp.create(3, 0)));
  StatsTestUtil.assertAggregationMapEquals(
      viewData1.getAggregationMap(),
      ImmutableMap.of(
          Arrays.asList(VALUE),
          StatsTestUtil.createAggregationData(DISTRIBUTION, measure1, value1)),
      EPSILON);
  assertThat(viewData2.getWindowData())
      .isEqualTo(CumulativeData.create(Timestamp.create(2, 0), Timestamp.create(4, 0)));
  StatsTestUtil.assertAggregationMapEquals(
      viewData2.getAggregationMap(),
      ImmutableMap.of(
          Arrays.asList(VALUE),
          StatsTestUtil.createAggregationData(DISTRIBUTION, measure2, value2)),
      EPSILON);
}
 

private static MeasureMap putToMeasureMap(MeasureMap measureMap, Measure measure, double value) {
  if (measure instanceof MeasureDouble) {
    return measureMap.put((MeasureDouble) measure, value);
  } else if (measure instanceof MeasureLong) {
    return measureMap.put((MeasureLong) measure, Math.round(value));
  } else {
    // Future measures.
    throw new AssertionError();
  }
}
 

private static void groupViewsByDirectoriesAndGetMeasures(
    Set<View> views, TreeNode root, Map<String, Measure> measures, Set<View> cachedViews) {
  for (View view : views) {
    if (cachedViews.contains(view)) {
      continue;
    }
    cachedViews.add(view);

    List<String> dirs = PATH_SPLITTER.splitToList(view.getName().asString());
    TreeNode node = root;
    for (int i = 0; i < dirs.size(); i++) {
      if (node == null) {
        break; // Should never happen. Work around the nullness checker.
      }
      String dir = dirs.get(i);
      if ("".equals(dir) && i == 0) {
        continue; // In case view name starts with a '/'.
      }
      node.views++;
      if (i != dirs.size() - 1) { // Non-leaf node (directory node)
        node.children.putIfAbsent(dir, new TreeNode());
        node = node.children.get(dir);
      } else { // Leaf node (view node)
        node.children.putIfAbsent(dir, new TreeNode(view.getName()));
      }
    }

    Measure measure = view.getMeasure();
    measures.putIfAbsent(measure.getName(), measure);
  }
}