/** Constructor. */
public BatchWrite(long maxBytes, int maxMessages, Duration maxDelay,
    PubsubMessageToTemplatedString batchKeyTemplate, Executor executor) {
  this.maxBytes = maxBytes;
  this.maxMessages = maxMessages;
  this.maxDelay = maxDelay;
  this.batchKeyTemplate = batchKeyTemplate;
  this.executor = executor;

  // create OpenCensus measures with a class name prefix
  final String shortClassName = this.getClass().getName().replaceAll(".*[.]", "");
  final String batchType = shortClassName.replace('$', '_').toLowerCase();
  totalBytes = MeasureLong.create(batchType + "_total_bytes",
      "The number of bytes received in " + shortClassName, "B");
  totalMessages = MeasureLong.create(batchType + "_total_messages",
      "The number of messages received in " + shortClassName, "1");
  batchCount = MeasureLong.create(batchType + "_batch_count",
      "The number of batches closed in " + shortClassName, "1");
  batchBytes = MeasureLong.create(batchType + "_batch_bytes",
      "Distribution of the number of bytes in a batch in " + shortClassName, "B");
  batchMessages = MeasureLong.create(batchType + "_batch_messages",
      "Distribution of the number of messages in a batch in " + shortClassName, "1");
  batchDelay = MeasureLong.create(batchType + "_batch_delay",
      "Distribution of the number of milliseconds a batch waited for messages in "
          + shortClassName,
      "ms");
}
 

/**
 * Register a view for every measure.
 *
 * <p>If this is not called, e.g. during unit tests, recorded values will not be exported.
 */
public Function<InputT, CompletableFuture<Void>> withOpenCensusMetrics() {
  final ViewManager viewManager = Stats.getViewManager();
  ImmutableMap.<MeasureLong, Aggregation>builder().put(batchCount, COUNT_AGG)
      .put(batchBytes, BATCH_BYTES_AGG).put(batchMessages, BATCH_MESSAGES_AGG)
      .put(batchDelay, BATCH_DELAY_AGG).put(totalBytes, SUM_AGG).put(totalMessages, SUM_AGG)
      .build()
      .forEach((measure, aggregation) -> viewManager
          .registerView(View.create(View.Name.create(measure.getName()), measure.getDescription(),
              measure, aggregation, ImmutableList.of())));
  return this;
}
 

/**
 * Register a view for every measure.
 *
 * <p>If this is not called, e.g. during unit tests, recorded values will not be exported.
 */
private static void setupOpenCensus() {
  ViewManager viewManager = Stats.getViewManager();
  for (MeasureLong measure : ImmutableList.of(COERCED_TO_INT, NOT_COERCED_TO_INT,
      NOT_COERCED_TO_BOOL)) {
    viewManager.registerView(View.create(Name.create(measure.getName()),
        measure.getDescription(), measure, COUNT_AGGREGATION, ImmutableList.of()));
  }
}
 

@Override
public MeasureMap put(MeasureLong measure, long value) {
  if (value < 0) {
    hasUnsupportedValues = true;
  }
  return this;
}
 

@Test
public void testMeasureLongComponents() {
  Measure measurement = Measure.MeasureLong.create("Bar", "The description of Bar", "1");
  assertThat(measurement.getName()).isEqualTo("Bar");
  assertThat(measurement.getDescription()).isEqualTo("The description of Bar");
  assertThat(measurement.getUnit()).isEqualTo("1");
}
 

@Test
public void testMeasureLongEquals() {
  new EqualsTester()
      .addEqualityGroup(
          Measure.MeasureLong.create("name", "description", "bit/s"),
          Measure.MeasureLong.create("name", "description", "bit/s"))
      .addEqualityGroup(Measure.MeasureLong.create("name", "description 2", "bit/s"))
      .testEquals();
}
 

@Override
public MeasureMapImpl put(MeasureLong measure, long value) {
  if (value < 0) {
    hasUnsupportedValues = true;
  }
  builder.put(measure, value);
  return this;
}
 

@Test
public void preventRegisteringDifferentMeasureWithSameName() {
  MeasureDouble measure1 = MeasureDouble.create("measure", "description", "1");
  MeasureLong measure2 = MeasureLong.create("measure", "description", "1");
  View view1 =
      View.create(
          VIEW_NAME, VIEW_DESCRIPTION, measure1, DISTRIBUTION, Arrays.asList(KEY), CUMULATIVE);
  View view2 =
      View.create(
          VIEW_NAME_2, VIEW_DESCRIPTION, measure2, DISTRIBUTION, Arrays.asList(KEY), CUMULATIVE);
  testFailedToRegisterView(
      view1, view2, "A different measure with the same name is already registered");
}
 

@Test
public void testRecord_MeasureTypeNotMatch() {
  testRecord_MeasureNotMatch(
      MeasureLong.create(MEASURE_NAME, "measure", MEASURE_UNIT),
      MeasureDouble.create(MEASURE_NAME, "measure", MEASURE_UNIT),
      10.0);
}
 

@Test
public void testMultipleViews_DifferentMeasureTypes() {
  testMultipleViews_DifferentMeasures(
      MeasureDouble.create(MEASURE_NAME, MEASURE_DESCRIPTION, MEASURE_UNIT),
      MeasureLong.create(MEASURE_NAME_2, MEASURE_DESCRIPTION, MEASURE_UNIT),
      1.1,
      5000);
}
 

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 MeasureLong createMeasureLong(String name, String desc) {
    MeasureLong msr = MeasureLong.create(name, desc == null ? name : desc, "");

    addView(msr);

    return msr;
}
 

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

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

private static void recordStat(MeasureLong ml, Long n) {
  TagContext empty = tagger.emptyBuilder().build();
  statsRecorder.newMeasureMap().put(ml, n).record(empty);
}
 

private static void recordTaggedStat(TagKey key, String value, MeasureLong ml, long n) {
  TagContext context = tagger.emptyBuilder().put(key, TagValue.create(value)).build();
  statsRecorder.newMeasureMap().put(ml, n).record(context);
}
 

@Override
public abstract MeasureLong getMeasure();
 

@Test
public void testMeasureDoubleIsNotEqualToMeasureLong() {
  assertThat(Measure.MeasureDouble.create("name", "description", "bit/s"))
      .isNotEqualTo(Measure.MeasureLong.create("name", "description", "bit/s"));
}
 

@Override
public MutableAggregation apply(MeasureLong arg) {
  return MutableSumLong.create();
}
 

@Override
public MutableAggregation apply(MeasureLong arg) {
  return MutableLastValueLong.create();
}
 

private static MeasureLong makeSimpleMeasureLong(String measure) {
  return Measure.MeasureLong.create(measure, measure + " description", "1");
}
 

@Override
public MeasureMap put(MeasureLong measure, long value) {
  return this;
}
 

private static void recordStat(MeasureLong ml, Long n) {
  TagContext empty = tagger.emptyBuilder().build();
  statsRecorder.newMeasureMap().put(ml, n).record(empty);
}
 

private static void recordTaggedStat(TagKey key, String value, MeasureLong ml, long n) {
  TagContext context = tagger.emptyBuilder().put(key, TagValue.create(value)).build();
  statsRecorder.newMeasureMap().put(ml, n).record(context);
}
 

public void init() {
  // Is this actually needed, not sure. Better safe than sorry!
  measures = new ConcurrentHashMap<String, MeasureLong>();
}
 

public void sendMetric(Metric metric) {
  try {
    String metricName = getOutputMetricName(metric);
    MeasureLong measure = measures.get(metricName);

    if (measure == null) {
      if (measures.size() > maxViews) {
        throw new RuntimeException("maxViews exceeded. " +
            "Please increase in configuration or decrease number of metrics.");
      }

      measure = MeasureLong.create("Events", "Number of Events", "1");
      measures.put(metricName, measure);

      // Stackdriver expects each metric to have the same set of tags so metrics
      // missing tags will be rejected. NB by default stackdriver will create
      // the metricDescription based on the first metric received so be consistant
      // from the start.
      final List<TagKey> columns = new ArrayList<TagKey>(metric.getTags().size());
      metric.getTags().keySet().forEach(tagName -> {
          columns.add(TagKey.create(sanitizeName(tagName)));
      });
      final View view =
          View.create(
              Name.create(metricName),
              metricName,
              measure,
              Sum.create(),
              columns);

      Stats.getViewManager().registerView(view);
    }
    final TagContextBuilder builder = tagger.emptyBuilder();
    metric.getTags().forEach((k, v) -> {
        builder.putPropagating(TagKey.create(sanitizeName(k)), TagValue.create(v));
    });
    final TagContext context = builder.build();

    statsRecorder.newMeasureMap().put(measure, (long) metric.getValue()).record(context);
  } catch (Exception ex) {
    log.error("Couldn't send metric %s", ex);
    throw ex;
  }
}
 

private void recordRealTimeMetric(TagContext ctx, MeasureLong measure, long value) {
  if (recordRealTimeMetrics) {
    MeasureMap measureMap = statsRecorder.newMeasureMap().put(measure, value);
    measureMap.record(ctx);
  }
}
 

/**
 * Constructs a new {@link MeasurementLong}.
 *
 * @since 0.8
 */
public static MeasurementLong create(MeasureLong measure, long value) {
  return new AutoValue_Measurement_MeasurementLong(measure, value);
}
 

/**
 * Associates the {@link MeasureLong} with the given value. Subsequent updates to the same {@link
 * MeasureLong} will overwrite the previous value.
 *
 * @param measure the {@link MeasureLong}
 * @param value the value to be associated with {@code measure}
 * @return this
 * @since 0.8
 */
public abstract MeasureMap put(MeasureLong measure, long value);
 

/**
 * Associates the {@link MeasureLong} with the given value. Subsequent updates to the same
 * {@link MeasureLong} will overwrite the previous value.
 *
 * @param measure the {@link MeasureLong}
 * @param value the value to be associated with {@code measure}
 * @return this
 */
Builder put(MeasureLong measure, long value) {
  measurements.add(Measurement.MeasurementLong.create(measure, value));
  return this;
}