Introduce an abstract super class for aggregators. (#898)

* Introduce an abstract super class for aggregators.
This is to hold the boilerplate logic and get it out of the specific implementations.

* tighten up the access modifiers and flip the assignable logic to be correct

sdk/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AbstractAggregator.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opentelemetry.sdk.metrics.aggregator;
+
+abstract class AbstractAggregator implements Aggregator {
+
+  @Override
+  public void mergeToAndReset(Aggregator other) {
+    if (!this.getClass().isAssignableFrom(other.getClass())) {
+      return;
+    }
+    doMergeAndReset(other);
+  }
+
+  /**
+   * Merges the current value into the given {@code aggregator} and resets the current value in this
+   * {@code Aggregator}.
+   *
+   * <p>If this method is called, you can assume that the passed in aggregator can be cast to your
+   * self-type.
+   *
+   * @param aggregator The aggregator to merge with.
+   */
+  abstract void doMergeAndReset(Aggregator aggregator);
+
+  @Override
+  public void recordLong(long value) {
+    throw new UnsupportedOperationException("This Aggregator does not support long values");
+  }
+
+  @Override
+  public void recordDouble(double value) {
+    throw new UnsupportedOperationException("This Aggregator does not support double values");
+  }
+}

sdk/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleSumAggregator.java

@@ -21,7 +21,8 @@
 import io.opentelemetry.sdk.metrics.data.MetricData.Point;
 import java.util.Map;
 
-public final class DoubleSumAggregator implements Aggregator {
+public final class DoubleSumAggregator extends AbstractAggregator {
+
   private static final AggregatorFactory AGGREGATOR_FACTORY =
       new AggregatorFactory() {
         @Override
@@ -38,11 +39,7 @@ public static AggregatorFactory getFactory() {
   }
 
   @Override
-  public void mergeToAndReset(Aggregator aggregator) {
-    if (!(aggregator instanceof DoubleSumAggregator)) {
-      return;
-    }
-
+  void doMergeAndReset(Aggregator aggregator) {
     DoubleSumAggregator other = (DoubleSumAggregator) aggregator;
     other.current.getAndAdd(this.current.getAndSet(0));
   }
@@ -52,15 +49,8 @@ public Point toPoint(long startEpochNanos, long epochNanos, Map<String, String>
     return DoublePoint.create(startEpochNanos, epochNanos, labels, current.get());
   }
 
-  @Override
-  public void recordLong(long value) {
-    throw new UnsupportedOperationException("This is a DoubleSumAggregator");
-  }
-
   @Override
   public void recordDouble(double value) {
     current.getAndAdd(value);
   }
-
-  DoubleSumAggregator() {}
 }

sdk/src/main/java/io/opentelemetry/sdk/metrics/aggregator/LongSumAggregator.java

@@ -21,7 +21,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
-public final class LongSumAggregator implements Aggregator {
+public final class LongSumAggregator extends AbstractAggregator {
   private static final AggregatorFactory AGGREGATOR_FACTORY =
       new AggregatorFactory() {
         @Override
@@ -38,11 +38,7 @@ public static AggregatorFactory getFactory() {
   }
 
   @Override
-  public void mergeToAndReset(Aggregator aggregator) {
-    if (!(aggregator instanceof LongSumAggregator)) {
-      return;
-    }
-
+  void doMergeAndReset(Aggregator aggregator) {
     LongSumAggregator other = (LongSumAggregator) aggregator;
     other.current.getAndAdd(this.current.getAndSet(0));
   }
@@ -52,15 +48,8 @@ public Point toPoint(long startEpochNanos, long epochNanos, Map<String, String>
     return LongPoint.create(startEpochNanos, epochNanos, labels, current.get());
   }
 
-  @Override
-  public void recordDouble(double value) {
-    throw new UnsupportedOperationException("This Aggregator does not support double values");
-  }
-
   @Override
   public void recordLong(long value) {
     current.getAndAdd(value);
   }
-
-  LongSumAggregator() {}
 }