Non-blocking: Improve non-blokcing handling

jknack · jknack · commit c5139cd94d45 · 2023-03-21T10:38:38.000-03:00
- Assume app is non-blocking while running in `event loop` mode
- Add fallback rendering to non-blocking filter
- Adjust handlers to return ctx
- Fix test supports
diff --git a/jooby/src/main/java/io/jooby/internal/Pipeline.java b/jooby/src/main/java/io/jooby/internal/Pipeline.java
@@ -8,6 +8,7 @@
 import static io.jooby.ReactiveSupport.concurrent;
 import static io.jooby.internal.handler.DefaultHandler.DEFAULT;
 import static io.jooby.internal.handler.DetachHandler.DETACH;
+import static io.jooby.internal.handler.SendDirect.DIRECT;
 import static io.jooby.internal.handler.WorkerHandler.WORKER;
 
 import java.lang.reflect.Type;
@@ -24,7 +25,6 @@
 import io.jooby.Route.Handler;
 import io.jooby.internal.handler.DispatchHandler;
 import io.jooby.internal.handler.PostDispatchInitializerHandler;
-import io.jooby.internal.handler.SendDirect;
 
 public class Pipeline {
 
@@ -35,6 +35,7 @@ public static Handler build(
       ContextInitializer initializer,
       Set<ResultHandler> responseHandler) {
     // Set default wrapper and blocking mode
+    route.setReactive(route.isReactive() || isDefaultReactive(executor, mode));
     Route.Filter wrapper = route.isReactive() ? DETACH : DEFAULT;
 
     /** Return type is set by annotation processor, or manually per lambda route: */
@@ -43,11 +44,10 @@ public static Handler build(
       Class<?> type = Reified.rawType(returnType);
       /** Context: */
       if (Context.class.isAssignableFrom(type)) {
-        if (executor == null && mode == ExecutionMode.EVENT_LOOP) {
-          route.setReactive(true);
+        if (route.isReactive()) {
           wrapper = DETACH;
         } else {
-          wrapper = SendDirect.DIRECT;
+          wrapper = DIRECT;
         }
       } else if (CompletionStage.class.isAssignableFrom(type)
           || Flow.Publisher.class.isAssignableFrom(type)) {
@@ -86,6 +86,10 @@ public static Handler build(
         mode, executor, decorate(initializer, wrapper.then(pipeline)), route.isReactive());
   }
 
+  private static boolean isDefaultReactive(Executor executor, ExecutionMode mode) {
+    return executor == null && mode == ExecutionMode.EVENT_LOOP;
+  }
+
   private static Handler decorate(ContextInitializer initializer, Handler handler) {
     return initializer == null
         ? handler
diff --git a/jooby/src/main/java/io/jooby/internal/handler/ConcurrentHandler.java b/jooby/src/main/java/io/jooby/internal/handler/ConcurrentHandler.java
@@ -23,8 +23,10 @@ public Route.Handler apply(@NonNull Route.Handler next) {
       Object result = next.apply(ctx);
       if (result instanceof Flow.Publisher publisher) {
         publisher.subscribe(newSubscriber(ctx));
+        // Return context to mark as handled
+        return ctx;
       } else if (result instanceof CompletionStage future) {
-        return future.whenComplete(
+        future.whenComplete(
             (value, x) -> {
               try {
                 Route.After after = ctx.getRoute().getAfter();
@@ -42,6 +44,8 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                 ctx.sendError(cause);
               }
             });
+        // Return context to mark as handled
+        return ctx;
       }
       return result;
     };
@@ -59,4 +63,9 @@ private Throwable unwrap(Throwable x) {
   public void setRoute(Route route) {
     route.setReactive(true);
   }
+
+  @Override
+  public String toString() {
+    return "concurrent";
+  }
 }
diff --git a/jooby/src/main/java/io/jooby/internal/handler/DefaultHandler.java b/jooby/src/main/java/io/jooby/internal/handler/DefaultHandler.java
@@ -30,4 +30,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {
       }
     };
   }
+
+  @Override
+  public String toString() {
+    return "default";
+  }
 }
diff --git a/jooby/src/main/java/io/jooby/internal/handler/DetachHandler.java b/jooby/src/main/java/io/jooby/internal/handler/DetachHandler.java
@@ -16,8 +16,24 @@ private DetachHandler() {}
 
   @NonNull @Override
   public Route.Handler apply(@NonNull Route.Handler next) {
-    return ctx -> {
-      return ctx.detach(next);
-    };
+    return context ->
+        context.detach(
+            (ctx -> {
+              try {
+                Object value = next.apply(ctx);
+                if (value != ctx && !ctx.isResponseStarted()) {
+                  ctx.render(value);
+                }
+                return value;
+              } catch (Throwable cause) {
+                ctx.sendError(cause);
+                return cause;
+              }
+            }));
+  }
+
+  @Override
+  public String toString() {
+    return "detach";
   }
 }
diff --git a/jooby/src/main/java/io/jooby/internal/handler/DispatchHandler.java b/jooby/src/main/java/io/jooby/internal/handler/DispatchHandler.java
@@ -30,4 +30,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {
               }
             });
   }
+
+  @Override
+  public String toString() {
+    return "dispatch";
+  }
 }
diff --git a/jooby/src/main/java/io/jooby/internal/handler/SendDirect.java b/jooby/src/main/java/io/jooby/internal/handler/SendDirect.java
@@ -26,4 +26,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {
       }
     };
   }
+
+  @Override
+  public String toString() {
+    return "direct";
+  }
 }
diff --git a/jooby/src/main/java/io/jooby/internal/handler/WorkerHandler.java b/jooby/src/main/java/io/jooby/internal/handler/WorkerHandler.java
@@ -25,4 +25,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {
               }
             });
   }
+
+  @Override
+  public String toString() {
+    return "worker";
+  }
 }
diff --git a/jooby/src/main/kotlin/io/jooby/CoroutineRouter.kt b/jooby/src/main/kotlin/io/jooby/CoroutineRouter.kt
@@ -74,10 +74,14 @@ class CoroutineRouter(val coroutineStart: CoroutineStart, val router: Router) {
         launch(handlerContext) {
           val result = handler(handlerContext)
           ctx.route.after?.apply(ctx, result, null)
-          if (result != ctx) {
+          if (result != ctx && !ctx.isResponseStarted) {
             ctx.render(result)
           }
+          // Return context to mark as handled
+          ctx
         }
+        // Return context to mark as handled
+        ctx
       }
       .setHandle(handler)
 
diff --git a/modules/jooby-mutiny/src/main/java/io/jooby/mutiny/Mutiny.java b/modules/jooby-mutiny/src/main/java/io/jooby/mutiny/Mutiny.java
@@ -60,8 +60,12 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                         // send error:
                         ctx.sendError((Throwable) failure);
                       });
+              // Return context to mark as handled
+              return ctx;
             } else if (result instanceof Multi multi) {
               multi.subscribe(newSubscriber(ctx));
+              // Return context to mark as handled
+              return ctx;
             }
             return result;
           };
diff --git a/modules/jooby-reactor/src/main/java/io/jooby/reactor/Reactor.java b/modules/jooby-reactor/src/main/java/io/jooby/reactor/Reactor.java
@@ -48,6 +48,8 @@ public Route.Handler apply(@NonNull Route.Handler next) {
             Object result = next.apply(ctx);
             if (result instanceof Flux flux) {
               flux.subscribe(toSubscriber(newSubscriber(ctx)));
+              // Return context to mark as handled
+              return ctx;
             } else if (result instanceof Mono mono) {
               mono.subscribe(
                   value -> {
@@ -62,6 +64,8 @@ public Route.Handler apply(@NonNull Route.Handler next) {
                     // send error:
                     ctx.sendError((Throwable) failure);
                   });
+              // Return context to mark as handled
+              return ctx;
             }
             return result;
           };
diff --git a/modules/jooby-rxjava3/src/main/java/io/jooby/rxjava3/Reactivex.java b/modules/jooby-rxjava3/src/main/java/io/jooby/rxjava3/Reactivex.java
@@ -36,12 +36,20 @@ public Route.Handler apply(@NonNull Route.Handler next) {
             Object result = next.apply(ctx);
             if (result instanceof Flowable flow) {
               flow.subscribe(toSubscriber(newSubscriber(ctx)));
+              // Return context to mark as handled
+              return ctx;
             } else if (result instanceof Single single) {
               single.subscribe(new RxSubscriber(ctx));
+              // Return context to mark as handled
+              return ctx;
             } else if (result instanceof Observable observable) {
               observable.subscribe(new RxObserver(newSubscriber(ctx)));
+              // Return context to mark as handled
+              return ctx;
             } else if (result instanceof Maybe maybe) {
               maybe.subscribe(new RxSubscriber(ctx));
+              // Return context to mark as handled
+              return ctx;
             }
             return result;
           };
diff --git a/modules/jooby-test/src/main/java/io/jooby/test/MockContext.java b/modules/jooby-test/src/main/java/io/jooby/test/MockContext.java
@@ -535,7 +535,6 @@ public MockContext render(@NonNull Object result) {
    * @return Mock response.
    */
   @NonNull public MockResponse getResponse() {
-    responseStarted = true;
     response.setHeaders(responseHeaders);
     return response;
   }
diff --git a/tests/src/test/java/io/jooby/test/FeaturedTest.java b/tests/src/test/java/io/jooby/test/FeaturedTest.java
@@ -171,28 +171,26 @@ public void sayHiFromIO(ServerTestRunner runner) {
     runner
         .define(
             app -> {
-              app.get("/", ctx -> "Hello World!");
+              app.get("/", ctx -> "Non-blocking:" + ctx.getRoute().isReactive());
               app.dispatch(
                   () -> {
-                    app.get("/worker", ctx -> "Hello World!");
+                    app.get("/worker", ctx -> "Non-blocking:" + ctx.getRoute().isReactive());
                   });
             })
         .ready(
             client -> {
               client.get(
                   "/",
                   rsp -> {
-                    assertEquals("Hello World!", rsp.body().string());
+                    assertEquals("Non-blocking:true", rsp.body().string());
                     assertEquals(200, rsp.code());
-                    assertEquals(12, rsp.body().contentLength());
                   });
 
               client.get(
                   "/worker",
                   rsp -> {
-                    assertEquals("Hello World!", rsp.body().string());
+                    assertEquals("Non-blocking:false", rsp.body().string());
                     assertEquals(200, rsp.code());
-                    assertEquals(12, rsp.body().contentLength());
                   });
 
               client.get(
@@ -242,28 +240,26 @@ public void sayHiFromWorker(ServerTestRunner runner) {
     runner
         .define(
             app -> {
-              app.get("/", ctx -> "Hello World!");
+              app.get("/", ctx -> "Non-blocking:" + ctx.getRoute().isReactive());
               app.dispatch(
                   () -> {
-                    app.get("/worker", ctx -> "Hello World!");
+                    app.get("/worker", ctx -> "Non-blocking:" + ctx.getRoute().isReactive());
                   });
             })
         .ready(
             client -> {
               client.get(
                   "/?foo=bar",
                   rsp -> {
-                    assertEquals("Hello World!", rsp.body().string());
+                    assertEquals("Non-blocking:false", rsp.body().string());
                     assertEquals(200, rsp.code());
-                    assertEquals(12, rsp.body().contentLength());
                   });
 
               client.get(
                   "/worker",
                   rsp -> {
-                    assertEquals("Hello World!", rsp.body().string());
+                    assertEquals("Non-blocking:false", rsp.body().string());
                     assertEquals(200, rsp.code());
-                    assertEquals(12, rsp.body().contentLength());
                   });
 
               client.get(
@@ -3606,18 +3602,20 @@ public void accessToFlashScopeFromTemplateEngine(ServerTestRunner runner) {
   }
 
   @ServerTest(executionMode = EVENT_LOOP)
-  @DisplayName("Context detaches when running in event-loop and returns a Context")
+  @DisplayName("Context detaches when running in event-loop")
   public void detachOnEventLoop(ServerTestRunner runner) {
     runner
         .define(
             app -> {
-              app.use(ReactiveSupport.concurrent());
               app.get(
                   "/detach",
                   ctx -> {
                     CompletableFuture.runAsync(
                         () -> {
-                          ctx.send(ctx.getRequestPath());
+                          ctx.send(
+                              ctx.getRequestPath()
+                                  + ". Non-blocking:"
+                                  + ctx.getRoute().isReactive());
                         });
                     return ctx;
                   });
@@ -3627,7 +3625,7 @@ public void detachOnEventLoop(ServerTestRunner runner) {
               client.get(
                   "/detach",
                   rsp -> {
-                    assertEquals("/detach", rsp.body().string().trim());
+                    assertEquals("/detach. Non-blocking:true", rsp.body().string().trim());
                   });
             });
   }

Original file line number	Diff line number	Diff line change
`@@ -30,4 +30,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {`
`30`	`30`	`}`
`31`	`31`	`};`
`32`	`32`	`}`
	`33`	`+`
	`34`	`+ @Override`
	`35`	`+ public String toString() {`
	`36`	`+ return "default";`
	`37`	`+ }`
`33`	`38`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,4 +26,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {`
`26`	`26`	`}`
`27`	`27`	`};`
`28`	`28`	`}`
	`29`	`+`
	`30`	`+ @Override`
	`31`	`+ public String toString() {`
	`32`	`+ return "direct";`
	`33`	`+ }`
`29`	`34`	`}`
Original file line number	Diff line number	Diff line change
`@@ -25,4 +25,9 @@ public Route.Handler apply(@NonNull Route.Handler next) {`
`25`	`25`	`}`
`26`	`26`	`});`
`27`	`27`	`}`
	`28`	`+`
	`29`	`+ @Override`
	`30`	`+ public String toString() {`
	`31`	`+ return "worker";`
	`32`	`+ }`
`28`	`33`	`}`
Original file line number	Diff line number	Diff line change
`@@ -74,10 +74,14 @@ class CoroutineRouter(val coroutineStart: CoroutineStart, val router: Router) {`
`74`	`74`	`launch(handlerContext) {`
`75`	`75`	`val result = handler(handlerContext)`
`76`	`76`	`ctx.route.after?.apply(ctx, result, null)`
`77`		`- if (result != ctx) {`
	`77`	`+ if (result != ctx && !ctx.isResponseStarted) {`
`78`	`78`	`ctx.render(result)`
`79`	`79`	`}`
	`80`	`+ // Return context to mark as handled`
	`81`	`+ ctx`
`80`	`82`	`}`
	`83`	`+ // Return context to mark as handled`
	`84`	`+ ctx`
`81`	`85`	`}`
`82`	`86`	`.setHandle(handler)`
`83`	`87`