Merge 34702f3 into d1b8033

ke4roh · Sep 27, 2019 · 1f2ebd6 · 1f2ebd6
2 parents d1b8033 + 34702f3
commit 1f2ebd6
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 10 deletions.
diff --git a/src/main/java/com/redhat/vertx/pipeline/Section.java b/src/main/java/com/redhat/vertx/pipeline/Section.java
@@ -83,13 +83,13 @@ public Completable init(Engine engine, JsonObject config) {
 
     public Maybe<Object> execute(JsonObject environment) {
         String documentId = environment.getJsonObject("doc").getString(Engine.DOC_UUID);
+        JsonObject stepdef = environment.getJsonObject("stepdef");
         return comboTechnique.apply(steps.stream().map(s -> s.executeStep(documentId)).collect(Collectors.toList()))
                 .doOnSubscribe(s -> publishSectionEvent(documentId, EventBusMessage.SECTION_STARTED))
                 .doOnComplete(() -> publishSectionEvent(documentId, EventBusMessage.SECTION_COMPLETED))
                 .doOnError(t -> publishSectionEvent(documentId, EventBusMessage.SECTION_ERRORED))
-                .lastElement()
-                .filter(r -> environment.getJsonObject("stepdef").getString("return",null) != null)
-                .map(r -> environment.getJsonObject("stepdef").getString("return"));
+                .ignoreElements()
+                .toMaybe();
     }
 
     private void publishSectionEvent(String documentId, String message) {

diff --git a/src/main/java/com/redhat/vertx/pipeline/StepExecutor.java b/src/main/java/com/redhat/vertx/pipeline/StepExecutor.java
@@ -2,7 +2,9 @@
 
 import com.redhat.vertx.Engine;
 import com.redhat.vertx.pipeline.json.TemplatedJsonObject;
+import io.reactivex.Completable;
 import io.reactivex.Maybe;
+import io.reactivex.Single;
 import io.vertx.core.json.JsonObject;
 
 import java.time.Duration;
@@ -33,14 +35,23 @@ private Maybe<Object> execute0(String docId) {
             return Maybe.empty();
         }
         String register = stepdef.getString("register");
+        boolean returnSomething = stepdef.containsKey("return");
 
         Maybe<Object> result = step.execute(stepEnvironment)
-                .filter(r -> register != null)
-                .flatMap(r ->
-                        engine
-                                .updateDocument(docId, new JsonObject().put(register,r))
-                                .toSingle(() -> r)
-                                .toMaybe()
+                .filter(r -> register != null || returnSomething )
+                .flatMap(r -> {
+                            Completable updateCompletable = register == null?
+                                    Completable.complete() :
+                                    engine.updateDocument(docId, new JsonObject().put(register, r));
+                            Maybe<Object> returnMaybe = returnSomething ?
+                                    Maybe.just(stepdef.getValue("return")) :
+                                    Maybe.empty();
+                            return updateCompletable.andThen(returnMaybe);
+                        },
+                        Maybe::error,
+                        () -> returnSomething ?
+                                Maybe.just(stepdef.getValue("return")) :
+                                Maybe.empty()
                 );
 
         Duration timeout = parseDuration(stepdef.getString("timeout", null));

diff --git a/src/test/java/com/redhat/vertx/pipeline/PipelineIntegrationTest.java b/src/test/java/com/redhat/vertx/pipeline/PipelineIntegrationTest.java
@@ -122,7 +122,6 @@ public void testEmptySection(Vertx vertx, VertxTestContext testContext) throws E
     }
 
     @Test
-    @Disabled
     public void returnASpecificValue(Vertx vertx, VertxTestContext testContext) throws Exception {
         Engine e = new Engine(ResourceUtils.fileContentsFromResource("com/redhat/vertx/pipeline/return-a-specific-value-pipeline.yaml"));
         vertx.rxDeployVerticle(e).blockingGet();