add helpers for working with Streams

For test cases it is sometimes useful to check aggregated
values rather than individual counters. The most common
example is looking at the overall number of increments for
a counter with a given name. The recommendation was to use
streams, but that can get a bit cumbersome:

```java
long sum = StreamSupport.stream(registry.spliterator(), false)
    .filter(m -> m instanceof Counter)        // Convert to counters
    .map(m -> (Counter) m)
    .filter(c -> "foo".equals(c.id().name())) // Restrict by name
    .mapToLong(Counter::count)                // Get value
    .reduce(0L, Long::sum)                    // Compute sum
```

This change adds some helpers to make this sort of use-case
a bit easier:

```java
long sum = registry.counters()
    .filter(Functions.nameEquals("foo"))      // Restrict by name
    .mapToLong(Counter::count)                // Get value
    .reduce(0L, Long::sum)                    // Compute sum
```

There are also some examples in the unit tests and comments
for the stream methods on Registry.

codequality/pmd.xml

@@ -37,6 +37,7 @@
     <exclude name="DuplicateImports"/>
     <exclude name="EmptyMethodInAbstractClassShouldBeAbstract"/>
     <exclude name="ExcessiveImports"/>
+    <exclude name="ExcessivePublicCount"/>
     <exclude name="FieldDeclarationsShouldBeAtStartOfClass"/>
     <exclude name="ForLoopsMustUseBraces"/>
     <exclude name="GodClass"/>

spectator-api/src/main/java/com/netflix/spectator/api/Functions.java

@@ -15,13 +15,15 @@
  */
 package com.netflix.spectator.api;
 
+import com.netflix.spectator.impl.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Predicate;
 import java.util.function.ToDoubleFunction;
 
 
@@ -108,4 +110,22 @@ public static ToDoubleFunction invokeMethod(final Method method) {
       }
     };
   }
+
+  /**
+   * Returns a predicate that matches if the {code}id.name(){code} value for the input meter
+   * is equal to {code}name{code}. Example of usage:
+   *
+   * <pre>
+   * long numberOfMatches = registry.stream().filter(Functions.nameEquals("foo")).count();
+   * </pre>
+   *
+   * @param name
+   *     The name to use for finding matching meter instances. Cannot be null.
+   * @return
+   *     A predicate function that can be used to filter a stream.
+   */
+  public static Predicate<Meter> nameEquals(String name) {
+    Preconditions.checkNotNull(name, "name");
+    return m -> name.equals(m.id().name());
+  }
 }

spectator-api/src/main/java/com/netflix/spectator/api/Registry.java

@@ -20,6 +20,8 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.function.ToDoubleFunction;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 /**
  * Registry to manage a set of meters.
@@ -617,4 +619,66 @@ default void methodValue(Id id, Object obj, String method) {
   default void methodValue(String name, Object obj, String method) {
     methodValue(createId(name), obj, method);
   }
+
+  /** Returns a stream of all registered meters. */
+  default Stream<Meter> stream() {
+    return StreamSupport.stream(spliterator(), false);
+  }
+
+  /**
+   * Returns a stream of all registered counters. This operation is mainly used for testing as
+   * a convenient way to get an aggregated value. For example, to generate a summary of all
+   * counters with name "foo":
+   *
+   * <pre>
+   * LongSummaryStatistics summary = r.counters()
+   *   .filter(Functions.nameEquals("foo"))
+   *   .collect(Collectors.summarizingLong(Counter::count));
+   * </pre>
+   */
+  default Stream<Counter> counters() {
+    return stream().filter(m -> m instanceof Counter).map(m -> (Counter) m);
+  }
+
+  /**
+   * Returns a stream of all registered distribution summaries. This operation is mainly used for
+   * testing as a convenient way to get an aggregated value. For example, to generate a summary of
+   * the counts and total amounts for all distribution summaries with name "foo":
+   *
+   * <pre>
+   * LongSummaryStatistics countSummary = r.distributionSummaries()
+   *   .filter(Functions.nameEquals("foo"))
+   *   .collect(Collectors.summarizingLong(DistributionSummary::count));
+   *
+   * LongSummaryStatistics totalSummary = r.distributionSummaries()
+   *   .filter(Functions.nameEquals("foo"))
+   *   .collect(Collectors.summarizingLong(DistributionSummary::totalAmount));
+   *
+   * double avgAmount = (double) totalSummary.getSum() / countSummary.getSum();
+   * </pre>
+   */
+  default Stream<DistributionSummary> distributionSummaries() {
+    return stream().filter(m -> m instanceof DistributionSummary).map(m -> (DistributionSummary) m);
+  }
+
+  /**
+   * Returns a stream of all registered timers. This operation is mainly used for testing as a
+   * convenient way to get an aggregated value. For example, to generate a summary of
+   * the counts and total amounts for all timers with name "foo":
+   *
+   * <pre>
+   * LongSummaryStatistics countSummary = r.timers()
+   *   .filter(Functions.nameEquals("foo"))
+   *   .collect(Collectors.summarizingLong(Timer::count));
+   *
+   * LongSummaryStatistics totalSummary = r.timers()
+   *   .filter(Functions.nameEquals("foo"))
+   *   .collect(Collectors.summarizingLong(Timer::totalTime));
+   *
+   * double avgTime = (double) totalSummary.getSum() / countSummary.getSum();
+   * </pre>
+   */
+  default Stream<Timer> timers() {
+    return stream().filter(m -> m instanceof Timer).map(m -> (Timer) m);
+  }
 }

spectator-api/src/test/java/com/netflix/spectator/api/ExtendedRegistryTest.java

@@ -22,10 +22,13 @@
 import org.junit.runners.JUnit4;
 
 import java.util.LinkedHashMap;
+import java.util.LongSummaryStatistics;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 @RunWith(JUnit4.class)
 public class ExtendedRegistryTest {
@@ -263,4 +266,74 @@ private int getValue() {
       return 42;
     }
   }
+
+  @Test
+  public void counters() {
+    Registry r = new DefaultRegistry();
+    r.counter("foo").increment();
+    r.counter("foo", "a", "1", "b", "2").increment();
+    r.counter("foo", "a", "1", "b", "3").increment(13L);
+    r.counter("foo", "a", "1", "b", "2").increment();
+    r.counter("bar", "a", "1", "b", "2").increment();
+
+    Assert.assertEquals(4, r.counters().count());
+    final LongSummaryStatistics summary = r.counters()
+        .filter(Functions.nameEquals("foo"))
+        .collect(Collectors.summarizingLong(Counter::count));
+    Assert.assertEquals(3L, summary.getCount());
+    Assert.assertEquals(16L, summary.getSum());
+    Assert.assertEquals(13L, summary.getMax());
+  }
+
+  @Test
+  public void timers() {
+    Registry r = new DefaultRegistry();
+    r.timer("foo").record(1L, TimeUnit.NANOSECONDS);
+    r.timer("foo", "a", "1", "b", "2").record(1L, TimeUnit.NANOSECONDS);
+    r.timer("foo", "a", "1", "b", "3").record(13L, TimeUnit.NANOSECONDS);
+    r.timer("foo", "a", "1", "b", "2").record(1L, TimeUnit.NANOSECONDS);
+    r.timer("bar", "a", "1", "b", "2").record(1L, TimeUnit.NANOSECONDS);
+
+    Assert.assertEquals(4, r.timers().count());
+
+    final LongSummaryStatistics countSummary = r.timers()
+        .filter(Functions.nameEquals("foo"))
+        .collect(Collectors.summarizingLong(Timer::count));
+    Assert.assertEquals(3L, countSummary.getCount());
+    Assert.assertEquals(4L, countSummary.getSum());
+    Assert.assertEquals(2L, countSummary.getMax());
+
+    final LongSummaryStatistics totalSummary = r.timers()
+        .filter(Functions.nameEquals("foo"))
+        .collect(Collectors.summarizingLong(Timer::totalTime));
+    Assert.assertEquals(3L, totalSummary.getCount());
+    Assert.assertEquals(16L, totalSummary.getSum());
+    Assert.assertEquals(13L, totalSummary.getMax());
+  }
+
+  @Test
+  public void distributionSummaries() {
+    Registry r = new DefaultRegistry();
+    r.distributionSummary("foo").record(1L);
+    r.distributionSummary("foo", "a", "1", "b", "2").record(1L);
+    r.distributionSummary("foo", "a", "1", "b", "3").record(13L);
+    r.distributionSummary("foo", "a", "1", "b", "2").record(1L);
+    r.distributionSummary("bar", "a", "1", "b", "2").record(1L);
+
+    Assert.assertEquals(4, r.distributionSummaries().count());
+
+    final LongSummaryStatistics countSummary = r.distributionSummaries()
+        .filter(Functions.nameEquals("foo"))
+        .collect(Collectors.summarizingLong(DistributionSummary::count));
+    Assert.assertEquals(3L, countSummary.getCount());
+    Assert.assertEquals(4L, countSummary.getSum());
+    Assert.assertEquals(2L, countSummary.getMax());
+
+    final LongSummaryStatistics totalSummary = r.distributionSummaries()
+        .filter(Functions.nameEquals("foo"))
+        .collect(Collectors.summarizingLong(DistributionSummary::totalAmount));
+    Assert.assertEquals(3L, totalSummary.getCount());
+    Assert.assertEquals(16L, totalSummary.getSum());
+    Assert.assertEquals(13L, totalSummary.getMax());
+  }
 }