Merge pull request #7 from jroper/tck-test-overhaul

Updated to newer TCK and fixed a number of bugs

Author: jroper

akka/src/main/java/com/lightbend/microprofile/reactive/streams/akka/AkkaEngine.java

@@ -197,7 +197,8 @@ else if (size == 1) {
       return Source.from(stage.getElements());
     });
     addSourceStage(Stage.PublisherStage.class, stage -> Source.fromPublisher(stage.getRsPublisher()));
-    addSourceStage(Stage.Concat.class, stage -> buildSource(stage.getFirst()).concat(buildSource(stage.getSecond())));
+    addSourceStage(Stage.Concat.class, stage -> buildSource(stage.getFirst())
+        .concat(buildSource(stage.getSecond())));
     addSourceStage(Stage.Failed.class, stage -> Source.failed(stage.getError()));
 
     // Flows
@@ -281,35 +282,38 @@ public akka.stream.Graph<SourceShape<Object>, NotUsed> apply(Throwable x, boolea
     // Sinks
     addSinkStage(Stage.FindFirst.class, stage -> Sink.headOption());
     addSinkStage(Stage.Collect.class, stage -> {
-      Collector collector = stage.getCollector();
-      BiConsumer accumulator = collector.accumulator();
-      Object firstContainer = collector.supplier().get();
-      if (firstContainer == null) {
-        firstContainer = NULL;
-      }
-      Sink<Object, CompletionStage<Object>> sink = Sink.fold(firstContainer, (resultContainer, in) -> {
-        if (resultContainer == NULL) {
-          accumulator.accept(null, in);
-        }
-        else {
-          accumulator.accept(resultContainer, in);
-        }
-        return resultContainer;
-      });
-      if (collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH) && firstContainer != NULL) {
-        return sink;
-      }
-      else {
-        return sink.mapMaterializedValue(result -> result.thenApply(r -> {
-          if (r == NULL) {
-            return collector.finisher().apply(null);
-          }
-          else {
-            return collector.finisher().apply(r);
-          }
-        }));
-      }
-    });
+          Collector collector = stage.getCollector();
+          // Lazy inited sink so that exceptions thrown by supplier get propagated through completion stage, not directly.
+          return Sink.lazyInitAsync(() -> {
+            BiConsumer accumulator = collector.accumulator();
+            Object firstContainer = collector.supplier().get();
+            if (firstContainer == null) {
+              firstContainer = NULL;
+            }
+            return CompletableFuture.completedFuture(Sink.fold(firstContainer, (resultContainer, in) -> {
+              if (resultContainer == NULL) {
+                accumulator.accept(null, in);
+              } else {
+                accumulator.accept(resultContainer, in);
+              }
+              return resultContainer;
+            }));
+          }).mapMaterializedValue(asyncMaybeResult -> asyncMaybeResult.thenCompose(maybeResult -> {
+            CompletionStage<Object> resultContainer;
+            if (maybeResult.isPresent()) {
+              resultContainer = maybeResult.get();
+            } else {
+              resultContainer = CompletableFuture.completedFuture(collector.supplier().get());
+            }
+            return resultContainer.thenApply(container -> {
+              if (collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH) && resultContainer != NULL) {
+                return container;
+              } else {
+                return collector.finisher().apply(container);
+              }
+            });
+          }));
+        });
     addSinkStage(Stage.SubscriberStage.class, stage ->
         Flow.create()
             .viaMat(new TerminationWatcher(), Keep.right())

akka/src/test/java/com/lightbend/microprofile/reactive/streams/akka/AkkaReactiveStreamsTckTest.java

@@ -8,10 +8,7 @@
 import akka.stream.ActorMaterializer;
 import akka.stream.Materializer;
 import org.eclipse.microprofile.reactive.streams.spi.ReactiveStreamsEngine;
-import org.eclipse.microprofile.reactive.streams.tck.CancelStageVerification;
-import org.eclipse.microprofile.reactive.streams.tck.FlatMapStageVerification;
 import org.eclipse.microprofile.reactive.streams.tck.ReactiveStreamsTck;
-import org.reactivestreams.tck.IdentityProcessorVerification;
 import org.reactivestreams.tck.TestEnvironment;
 import org.testng.annotations.AfterSuite;
 
@@ -40,9 +37,4 @@ protected AkkaEngine createEngine() {
     materializer = ActorMaterializer.create(system);
     return new AkkaEngine(materializer);
   }
-
-  @Override
-  protected boolean isEnabled(Object test) {
-    return true;
-  }
 }

bin/buildDeps.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 # This should be set to the commit hash that is being tracked. Needed even if TRACKING_PR is set.
-TRACKING_COMMIT="886a91d5"
+TRACKING_COMMIT="0d0c4f50"
 # To track a particular pull request, put it's number here, otherwise comment it out.
 # TRACKING_PR="67"
 

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/BuiltGraph.java

@@ -269,9 +269,6 @@ private Builder buildGraph(Graph graph, Shape shape) {
         }
       }
 
-      ports.addAll(builderPorts);
-      stages.addAll(builderStages);
-
       return this;
     }
 
@@ -284,8 +281,22 @@ private void verifyReady() {
       for (Port port : builderPorts) {
         port.verifyReady();
       }
+      ports.addAll(builderPorts);
+    }
+
+    /**
+     * Start the stages on this listener
+     */
+    private void startGraph() {
+      execute(() -> {
+        for (GraphStage stage : builderStages) {
+          stages.add(stage);
+          stage.postStart();
+        }
+      });
     }
 
+
     private <T> SubStageInlet<T> inlet() {
       Objects.requireNonNull(lastInlet, "Not an inlet graph");
       assert result == null;
@@ -360,18 +371,8 @@ private void addStage(Stage stage, StageInlet inlet, Publisher publisher, StageO
           addStage(new OfStage(BuiltGraph.this, outlet,
               ((Stage.Of) stage).getElements()));
         } else if (stage instanceof Stage.Concat) {
-
-          // Use this builder to build each of the sub stages that are being concatenated as an inlet graph, and then
-          // capture the last inlet of each to pass to the concat stage.
-          buildGraph(((Stage.Concat) stage).getFirst(), Shape.INLET);
-          StageInlet firstInlet = lastInlet;
-          lastInlet = null;
-
-          buildGraph(((Stage.Concat) stage).getSecond(), Shape.INLET);
-          StageInlet secondInlet = lastInlet;
-          lastInlet = null;
-
-          addStage(new ConcatStage(BuiltGraph.this, firstInlet, secondInlet, outlet));
+          Stage.Concat concat = (Stage.Concat) stage;
+          addStage(new ConcatStage(BuiltGraph.this, buildSubInlet(concat.getFirst()), buildSubInlet(concat.getSecond()), outlet));
         } else if (stage instanceof Stage.PublisherStage) {
           addStage(new ConnectorStage(BuiltGraph.this, ((Stage.PublisherStage) stage).getRsPublisher(), subscriber));
         } else if (stage instanceof Stage.Failed) {
@@ -517,17 +518,6 @@ public void execute(Runnable command) {
     });
   }
 
-  /**
-   * Start the whole graph.
-   */
-  private void startGraph() {
-    execute(() -> {
-      for (GraphStage stage : stages) {
-        stage.postStart();
-      }
-    });
-  }
-
   private void streamFailure(Throwable error) {
     // todo handle better
     error.printStackTrace();
@@ -558,8 +548,6 @@ final class SubStageInlet<T> implements StageInlet<T> {
     private final List<GraphStage> subStages;
     private final List<Port> subStagePorts;
 
-    private boolean started = false;
-
     private SubStageInlet(StageInlet<T> delegate, List<GraphStage> subStages, List<Port> subStagePorts) {
       this.delegate = delegate;
       this.subStages = subStages;
@@ -568,20 +556,24 @@ private SubStageInlet(StageInlet<T> delegate, List<GraphStage> subStages, List<P
 
     void start() {
       subStagePorts.forEach(Port::verifyReady);
-      started = true;
-      subStages.forEach(GraphStage::postStart);
+      ports.addAll(subStagePorts);
+      for (GraphStage stage: subStages) {
+        stages.add(stage);
+        stage.postStart();
+      }
     }
 
     private void shutdown() {
-      stages.removeAll(subStages);
-      ports.removeAll(subStagePorts);
+      // Do it in a signal, this ensures that if shutdown happens while something is iterating through
+      // the ports, we don't get a concurrent modification exception.
+      enqueueSignal(() -> {
+        stages.removeAll(subStages);
+        ports.removeAll(subStagePorts);
+      });
     }
 
     @Override
     public void pull() {
-      if (!started) {
-        throw new IllegalStateException("Pull before the sub stream has been started.");
-      }
       delegate.pull();
     }
 
@@ -602,17 +594,11 @@ public boolean isClosed() {
 
     @Override
     public void cancel() {
-      if (!started) {
-        throw new IllegalStateException("Cancel before the sub stream has been started.");
-      }
       delegate.cancel();
     }
 
     @Override
     public T grab() {
-      if (!started) {
-        throw new IllegalStateException("Grab before the sub stream has been started.");
-      }
       return delegate.grab();
     }
 

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/CollectStage.java

@@ -23,12 +23,12 @@ public CollectStage(BuiltGraph builtGraph, StageInlet<T> inlet,
     this.result = result;
     this.collector = collector;
 
-    container = collector.supplier().get();
     inlet.setListener(this);
   }
 
   @Override
   protected void postStart() {
+    container = collector.supplier().get();
     inlet.pull();
   }
 

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/ConcatStage.java

@@ -4,92 +4,79 @@
 
 package com.lightbend.microprofile.reactive.streams.zerodep;
 
-public class ConcatStage<T> extends GraphStage implements OutletListener {
+public class ConcatStage<T> extends GraphStage implements InletListener, OutletListener {
 
-  private final StageInlet<T> first;
-  private final StageInlet<T> second;
+  private final BuiltGraph.SubStageInlet<T> first;
+  private final BuiltGraph.SubStageInlet<T> second;
   private final StageOutlet<T> outlet;
 
-  private Throwable secondError;
-
-  public ConcatStage(BuiltGraph builtGraph, StageInlet<T> first, StageInlet<T> second, StageOutlet<T> outlet) {
+  public ConcatStage(BuiltGraph builtGraph, BuiltGraph.SubStageInlet<T> first, BuiltGraph.SubStageInlet<T> second, StageOutlet<T> outlet) {
     super(builtGraph);
     this.first = first;
     this.second = second;
     this.outlet = outlet;
 
-    first.setListener(new FirstInletListener());
-    second.setListener(new SecondInletListener());
+    first.setListener(this);
     outlet.setListener(this);
   }
 
+  @Override
+  protected void postStart() {
+    first.start();
+  }
+
   @Override
   public void onPull() {
-    if (first.isClosed()) {
-      second.pull();
-    } else {
-      first.pull();
-    }
+    first.pull();
   }
 
   @Override
   public void onDownstreamFinish() {
-    if (!first.isClosed()) {
-      first.cancel();
-    }
-    if (!second.isClosed()) {
-      second.cancel();
-    }
+    first.cancel();
+    // Start up second so we can shut it down, in case it holds any resources.
+    startAndCancelSecond();
   }
 
-  private class FirstInletListener implements InletListener {
-    @Override
-    public void onPush() {
-      outlet.push(first.grab());
-    }
+  @Override
+  public void onPush() {
+    outlet.push(first.grab());
+  }
 
-    @Override
-    public void onUpstreamFinish() {
-      if (second.isClosed()) {
-        if (secondError != null) {
-          outlet.fail(secondError);
-        } else {
-          outlet.complete();
-        }
-      } else if (outlet.isAvailable()) {
-        second.pull();
-      }
+  @Override
+  public void onUpstreamFinish() {
+    second.forwardTo(outlet);
+    outlet.forwardTo(second);
+    second.start();
+    if (outlet.isAvailable()) {
+      second.pull();
     }
+  }
 
-    @Override
-    public void onUpstreamFailure(Throwable error) {
-      outlet.fail(error);
-      if (!second.isClosed()) {
-        second.cancel();
-      }
-    }
+  @Override
+  public void onUpstreamFailure(Throwable error) {
+    outlet.fail(error);
+    startAndCancelSecond();
   }
 
-  private class SecondInletListener implements InletListener {
-    @Override
-    public void onPush() {
-      outlet.push(second.grab());
-    }
+  private void startAndCancelSecond() {
+    try {
+      second.setListener(new InletListener() {
+        @Override
+        public void onPush() {
+        }
 
-    @Override
-    public void onUpstreamFinish() {
-      if (first.isClosed()) {
-        outlet.complete();
-      }
-    }
+        @Override
+        public void onUpstreamFinish() {
+        }
 
-    @Override
-    public void onUpstreamFailure(Throwable error) {
-      if (first.isClosed()) {
-        outlet.fail(error);
-      } else {
-        secondError = error;
-      }
+        @Override
+        public void onUpstreamFailure(Throwable error) {
+        }
+      });
+      second.start();
+      second.cancel();
+    } catch (Exception e) {
+      // Ignore exceptions
     }
   }
 }