[WIP] [QUARKS-195] Metrics.counter needs TStream.peek(oplet)

api/topology/src/main/java/quarks/topology/TStream.java

@@ -31,6 +31,7 @@ Licensed to the Apache Software Foundation (ASF) under one
 import quarks.function.ToIntFunction;
 import quarks.function.UnaryOperator;
 import quarks.oplet.core.FanIn;
+import quarks.oplet.core.Peek;
 import quarks.oplet.core.Pipe;
 import quarks.oplet.core.Sink;
 
@@ -249,6 +250,18 @@ else if (WARNING.equals(lr.getLevel()))
      */
     TStream<T> peek(Consumer<T> peeker);
 
+    /**
+     * Declare a stream that contains the same contents as this stream while
+     * peeking at each element using {@code oplet}. <BR>
+     * For each tuple {@code t} on this stream, {@code oplet.peek(t)} will be
+     * called.
+     * 
+     * @param oplet
+     *            the {@link Peek} oplet.
+     * @return {@code this}
+     */
+    TStream<T> peek(Peek<T> oplet);
+
     /**
      * Sink (terminate) this stream using a function. For each tuple {@code t} on this stream
      * {@link Consumer#accept(Object) sinker.accept(t)} will be called. This is

spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java

@@ -144,9 +144,13 @@ public <E extends Enum<E>> EnumMap<E,TStream<T>> split(Class<E> enumClass, Funct
 
     @Override
     public TStream<T> peek(Consumer<T> peeker) {
-        peeker = Functions.synchronizedConsumer(peeker);
-        connector.peek(new Peek<T>(peeker));
-        return this;
+        return peek(new Peek<T>(Functions.synchronizedConsumer(peeker)));
+    }
+
+    @Override
+    public TStream<T> peek(quarks.oplet.core.Peek<T> peek) {
+      connector.peek(peek);
+      return this;
     }
 
     @Override
@@ -158,6 +162,8 @@ public TSink<T> sink(Sink<T> oplet) {
 
     @Override
     public <U> TStream<U> pipe(Pipe<T, U> pipe) {
+        if (pipe instanceof quarks.oplet.core.Peek<?>)
+          throw new IllegalArgumentException("Use peek() to add Peek oplets");
         return connectPipe(pipe);
     }
 

utils/metrics/src/main/java/quarks/metrics/Metrics.java

@@ -30,26 +30,34 @@ public class Metrics {
     /**
      * Increment a counter metric when peeking at each tuple.
      * 
+     * <P>
+     * Same as {@code stream.peek(new CounterOp<T>())}
+     * </P>
+     * 
      * @param <T>
      *            TStream tuple type
-     * @param stream to stream to instrument
-     * @return a {@link TStream} containing the input tuples
+     * @param stream the stream to monitor
+     * @return the {@code stream} argument
      */
     public static <T> TStream<T> counter(TStream<T> stream) {
-        return stream.pipe(new CounterOp<T>());
+        return stream.peek(new CounterOp<T>());
     }
 
     /**
      * Measure current tuple throughput and calculate one-, five-, and
      * fifteen-minute exponentially-weighted moving averages.
      * 
+     * <P>
+     * Same as {@code stream.peek(new RateMeter<T>())}
+     * </P>
+     * 
      * @param <T>
      *            TStream tuple type
-     * @param stream to stream to instrument
-     * @return a {@link TStream} containing the input tuples
+     * @param stream the stream to monitor
+     * @return the {@code stream} argument
      */
     public static <T> TStream<T> rateMeter(TStream<T> stream) {
-        return stream.pipe(new RateMeter<T>());
+        return stream.peek(new RateMeter<T>());
     }
 
     /**

utils/metrics/src/test/java/quarks/test/metrics/MetricsEverywhereTest.java

@@ -74,7 +74,7 @@ public void automaticMetricCleanup1() throws Exception {
         Topology t = newTopology();
         AtomicInteger n = new AtomicInteger(0);
         TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 10, TimeUnit.MILLISECONDS);
-        ints.pipe(new TestOplet<Integer>());
+        ints.peek(new TestOplet<Integer>());
 
         // Submit job
         Future<? extends Job> fj = getSubmitter().submit(t);
@@ -104,8 +104,8 @@ public void automaticMetricCleanup2() throws Exception {
         Topology t = newTopology();
         AtomicInteger n = new AtomicInteger(0);
         TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 10, TimeUnit.MILLISECONDS);
-        TStream<Integer> ints2 = ints.pipe(new TestOplet<Integer>());
-        ints2.pipe(new TestOplet<Integer>());
+        TStream<Integer> ints2 = ints.peek(new TestOplet<Integer>());
+        ints2.peek(new TestOplet<Integer>());
 
         // Submit job
         Future<? extends Job> fj = getSubmitter().submit(t);
@@ -154,6 +154,20 @@ public void metricsEverywhereSplit() throws Exception {
          * OP_0 -- OP_1(Split) ----- OP_4 (Sink)
          *                       \
          *                        -- OP_5 (Sink)
+         *                        
+         * Note, OP_2 (Counter) is a peek oplet and as such
+         * is part of the split[0] stream's peek-chain.  
+         * Metrics insertion only occurs at the end of a peek-chain,
+         * so there is NOT an injected metric between 
+         * OP_1(Split) -> OP_2 (Counter).
+         * 
+         * The net is metrics are injected where the "#" are:
+         * 
+         *                        -- OP_2 (Counter) -#- OP_3 (Sink)
+         *                       / 
+         * OP_0 -#- OP_1(Split) ---#- OP_4 (Sink)
+         *                       \
+         *                        -#- OP_5 (Sink)
          */
         Topology t = newTopology();
         Graph g = t.graph();