reactive: Reactive types must support side-effect handlers #3486

-  fix #3486

Author: jknack

jooby/src/main/java/io/jooby/internal/handler/ConcurrentHandler.java

@@ -41,7 +41,10 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                   Throwable exception = unwrap((Throwable) x);
                   ctx.sendError(exception);
                 } else {
-                  ctx.render(value);
+                  // See https://github.com/jooby-project/jooby/issues/3486
+                  if (!ctx.isResponseStarted() && value != ctx) {
+                    ctx.render(value);
+                  }
                 }
               } catch (Throwable cause) {
                 ctx.sendError(cause);

modules/jooby-mutiny/src/main/java/io/jooby/mutiny/Mutiny.java

@@ -55,8 +55,11 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                       value -> {
                         // fire after:
                         after(ctx, value, null);
-                        // render:
-                        ctx.render(value);
+                        // See https://github.com/jooby-project/jooby/issues/3486
+                        if (!ctx.isResponseStarted() && value != ctx) {
+                          // render:
+                          ctx.render(value);
+                        }
                       },
                       failure -> {
                         // fire after:

modules/jooby-reactor/src/main/java/io/jooby/reactor/Reactor.java

@@ -59,8 +59,12 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                   value -> {
                     // fire after:
                     after(ctx, value, null);
-                    // render:
-                    ctx.render(value);
+                    // See https://github.com/jooby-project/jooby/issues/3486
+                    if (!ctx.isResponseStarted() && value != ctx) {
+
+                      // render:
+                      ctx.render(value);
+                    }
                   },
                   failure -> {
                     // fire after:

modules/jooby-rxjava3/src/main/java/io/jooby/internal/rxjava3/RxSubscriber.java

@@ -18,12 +18,12 @@
 
 public class RxSubscriber implements MaybeObserver<Object>, SingleObserver<Object> {
 
-  private final Context context;
+  private final Context ctx;
 
   private Disposable subscription;
 
-  public RxSubscriber(Context context) {
-    this.context = context;
+  public RxSubscriber(Context ctx) {
+    this.ctx = ctx;
   }
 
   @Override
@@ -33,14 +33,17 @@ public void onSubscribe(Disposable d) {
 
   @Override
   public void onSuccess(Object value) {
-    after(context, value, null);
-    context.render(value);
+    after(ctx, value, null);
+    // See https://github.com/jooby-project/jooby/issues/3486
+    if (!ctx.isResponseStarted() && value != ctx) {
+      ctx.render(value);
+    }
   }
 
   @Override
   public void onError(Throwable x) {
-    after(context, null, unwrap(x));
-    context.sendError(x);
+    after(ctx, null, unwrap(x));
+    ctx.sendError(x);
     subscription.dispose();
   }
 
@@ -53,9 +56,9 @@ private Throwable unwrap(Throwable x) {
 
   @Override
   public void onComplete() {
-    if (!context.isResponseStarted()) {
+    if (!ctx.isResponseStarted()) {
       // assume it is a maybe response:
-      context.send(StatusCode.NOT_FOUND);
+      ctx.send(StatusCode.NOT_FOUND);
     }
     subscription.dispose();
   }

tests/src/test/java/io/jooby/i3486/Issue3486.java

@@ -0,0 +1,98 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.i3486;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.concurrent.CompletableFuture;
+
+import io.jooby.ReactiveSupport;
+import io.jooby.junit.ServerTest;
+import io.jooby.junit.ServerTestRunner;
+import io.jooby.mutiny.Mutiny;
+import io.jooby.reactor.Reactor;
+import io.jooby.rxjava3.Reactivex;
+import io.reactivex.rxjava3.core.Single;
+import io.smallrye.mutiny.Uni;
+import reactor.core.publisher.Mono;
+
+public class Issue3486 {
+  @ServerTest
+  public void reactiveShouldWorkWithSideEffectsHandler(ServerTestRunner runner) {
+    runner
+        .define(
+            app -> {
+              app.use(ReactiveSupport.concurrent());
+              app.use(Mutiny.mutiny());
+              app.use(Reactor.reactor());
+              app.use(Reactivex.rx());
+
+              app.get(
+                  "/3486/completablefuture",
+                  ctx ->
+                      CompletableFuture.supplyAsync(
+                          () -> {
+                            ctx.send("top page");
+                            return ctx;
+                          }));
+
+              app.get(
+                  "/3486/mutiny",
+                  ctx ->
+                      Uni.createFrom()
+                          .item(
+                              () -> {
+                                ctx.send("top page");
+                                return ctx;
+                              }));
+
+              app.get(
+                  "/3486/reactor",
+                  ctx ->
+                      Mono.fromSupplier(
+                          () -> {
+                            ctx.send("top page");
+                            return ctx;
+                          }));
+
+              app.get(
+                  "/3486/reactivex",
+                  ctx ->
+                      Single.fromSupplier(
+                          () -> {
+                            ctx.send("top page");
+                            return ctx;
+                          }));
+            })
+        .ready(
+            client -> {
+              client.get(
+                  "/3486/completablefuture",
+                  rsp -> {
+                    assertEquals(200, rsp.code());
+                    assertEquals("top page", rsp.body().string());
+                  });
+              client.get(
+                  "/3486/mutiny",
+                  rsp -> {
+                    assertEquals(200, rsp.code());
+                    assertEquals("top page", rsp.body().string());
+                  });
+              client.get(
+                  "/3486/reactor",
+                  rsp -> {
+                    assertEquals(200, rsp.code());
+                    assertEquals("top page", rsp.body().string());
+                  });
+              client.get(
+                  "/3486/reactivex",
+                  rsp -> {
+                    assertEquals(200, rsp.code());
+                    assertEquals("top page", rsp.body().string());
+                  });
+            });
+  }
+}