Use body handler to allow request buffering

Fixes quarkusio#5959 This change allows the body handler to be used for Undertow and RESTEasy standalone. When the request is fully buffered it can be consumed multiple times, which allows keycloak to also process it.
stuartwdouglas · Dec 6, 2019 · 3a81948 · 3a81948
1 parent 4abf92f
commit 3a81948
Show file tree

Hide file tree

Showing 13 changed files with 118 additions and 57 deletions.
diff --git a/bom/runtime/pom.xml b/bom/runtime/pom.xml
@@ -24,7 +24,7 @@
         <opentracing-concurrent.version>0.2.0</opentracing-concurrent.version>
         <opentracing-jdbc.version>0.0.12</opentracing-jdbc.version>
         <jaeger.version>0.34.0</jaeger.version>
-        <quarkus-http.version>3.0.0.Final</quarkus-http.version>
+        <quarkus-http.version>3.0.1.Final</quarkus-http.version>
         <jboss-servlet-api_4.0_spec.version>1.0.0.Final</jboss-servlet-api_4.0_spec.version>
         <microprofile-config-api.version>1.3</microprofile-config-api.version>
         <microprofile-context-propagation.version>1.0.1</microprofile-context-propagation.version>

diff --git a/...ent/src/main/java/io/quarkus/keycloak/pep/deployment/KeycloakPolicyEnforcerBuildStep.java b/...ent/src/main/java/io/quarkus/keycloak/pep/deployment/KeycloakPolicyEnforcerBuildStep.java
@@ -13,6 +13,7 @@
 import io.quarkus.oidc.OIDCException;
 import io.quarkus.oidc.runtime.OidcBuildTimeConfig;
 import io.quarkus.oidc.runtime.OidcConfig;
+import io.quarkus.vertx.http.deployment.RequireBodyHandlerBuildItem;
 
 public class KeycloakPolicyEnforcerBuildStep {
 
@@ -21,6 +22,12 @@ FeatureBuildItem featureBuildItem() {
         return new FeatureBuildItem(FeatureBuildItem.KEYCLOAK_AUTHORIZATION);
     }
 
+    @BuildStep
+    RequireBodyHandlerBuildItem requireBody() {
+        //TODO: this should only be produced if required, ask Pedro about config for this
+        return new RequireBodyHandlerBuildItem();
+    }
+
     @BuildStep
     public AdditionalBeanBuildItem beans(KeycloakPolicyEnforcerConfig config) {
         if (config.policyEnforcer.enable) {

diff --git a/...-authorization/runtime/src/main/java/io/quarkus/keycloak/pep/runtime/VertxHttpFacade.java b/...-authorization/runtime/src/main/java/io/quarkus/keycloak/pep/runtime/VertxHttpFacade.java
@@ -1,6 +1,5 @@
 package io.quarkus.keycloak.pep.runtime;
 
-import java.io.BufferedInputStream;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.InputStream;
@@ -11,8 +10,6 @@
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.security.cert.X509Certificate;
 
-import io.quarkus.vertx.http.runtime.VertxInputStream;
-import io.vertx.core.http.HttpHeaders;
 import org.keycloak.KeycloakSecurityContext;
 import org.keycloak.adapters.OIDCHttpFacade;
 import org.keycloak.adapters.spi.AuthenticationError;
@@ -25,8 +22,10 @@
 import io.quarkus.oidc.AccessTokenCredential;
 import io.quarkus.security.credential.TokenCredential;
 import io.quarkus.security.identity.SecurityIdentity;
+import io.quarkus.vertx.http.runtime.VertxInputStream;
 import io.quarkus.vertx.http.runtime.security.QuarkusHttpUser;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
 import io.vertx.core.http.impl.CookieImpl;
@@ -133,15 +132,13 @@ public InputStream getInputStream() {
             @Override
             public InputStream getInputStream(boolean buffered) {
                 try {
-                    if (routingContext.get("quarkus.request.inputstream") != null) {
-                        return routingContext.get("quarkus.request.inputstream");
+                    if (routingContext.getBody() != null) {
+                        return new ByteArrayInputStream(routingContext.getBody().getBytes());
                     }
-
-                    BufferedInputStream stream = new BufferedInputStream(new VertxInputStream(request));
-
-                    routingContext.put("quarkus.request.inputstream", stream);
-
-                    return stream;
+                    if (routingContext.request().isEnded()) {
+                        return new ByteArrayInputStream(new byte[0]);
+                    }
+                    return new VertxInputStream(request);
                 } catch (Exception e) {
                     throw new RuntimeException(e);
                 }

diff --git a/...asy/runtime/src/main/java/io/quarkus/resteasy/runtime/standalone/VertxRequestHandler.java b/...asy/runtime/src/main/java/io/quarkus/resteasy/runtime/standalone/VertxRequestHandler.java
@@ -1,13 +1,13 @@
 package io.quarkus.resteasy.runtime.standalone;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 
 import javax.enterprise.inject.Instance;
 import javax.enterprise.inject.spi.CDI;
 import javax.ws.rs.core.SecurityContext;
 
-import io.quarkus.vertx.http.runtime.VertxInputStream;
 import org.jboss.logging.Logger;
 import org.jboss.resteasy.core.ResteasyContext;
 import org.jboss.resteasy.core.SynchronousDispatcher;
@@ -20,6 +20,7 @@
 import io.quarkus.arc.runtime.BeanContainer;
 import io.quarkus.security.identity.CurrentIdentityAssociation;
 import io.quarkus.vertx.http.runtime.CurrentVertxRequest;
+import io.quarkus.vertx.http.runtime.VertxInputStream;
 import io.quarkus.vertx.http.runtime.security.QuarkusHttpUser;
 import io.vertx.core.Context;
 import io.vertx.core.Handler;
@@ -64,9 +65,8 @@ public void handle(RoutingContext request) {
         // otherwise request handlers may not get set up before request ends
         InputStream is;
         try {
-            if (request.get("quarkus.request.inputstream") != null) {
-                is = request.get("quarkus.request.inputstream");
-                is.mark(0);
+            if (request.getBody() != null) {
+                is = new ByteArrayInputStream(request.getBody().getBytes());
             } else {
                 is = new VertxInputStream(request.request());
             }

diff --git a/...ndertow/runtime/src/main/java/io/quarkus/undertow/runtime/UndertowDeploymentRecorder.java b/...ndertow/runtime/src/main/java/io/quarkus/undertow/runtime/UndertowDeploymentRecorder.java
@@ -353,7 +353,8 @@ public void run() {
         return new Handler<RoutingContext>() {
             @Override
             public void handle(RoutingContext event) {
-                VertxHttpExchange exchange = new VertxHttpExchange(event.request(), allocator, executorService, event);
+                VertxHttpExchange exchange = new VertxHttpExchange(event.request(), allocator, executorService, event,
+                        event.getBody());
                 Optional<MemorySize> maxBodySize = httpConfiguration.limits.maxBodySize;
                 if (maxBodySize.isPresent()) {
                     exchange.setMaxEntitySize(maxBodySize.get().asLongValue());

diff --git a/...-http/deployment/src/main/java/io/quarkus/vertx/http/deployment/BodyHandlerBuildItem.java b/...-http/deployment/src/main/java/io/quarkus/vertx/http/deployment/BodyHandlerBuildItem.java
@@ -0,0 +1,17 @@
+package io.quarkus.vertx.http.deployment;
+
+import io.quarkus.builder.item.SimpleBuildItem;
+import io.vertx.core.Handler;
+import io.vertx.ext.web.RoutingContext;
+
+public final class BodyHandlerBuildItem extends SimpleBuildItem {
+    private final Handler<RoutingContext> handler;
+
+    public BodyHandlerBuildItem(Handler<RoutingContext> handler) {
+        this.handler = handler;
+    }
+
+    public Handler<RoutingContext> getHandler() {
+        return handler;
+    }
+}
diff --git a/...eployment/src/main/java/io/quarkus/vertx/http/deployment/RequireBodyHandlerBuildItem.java b/...eployment/src/main/java/io/quarkus/vertx/http/deployment/RequireBodyHandlerBuildItem.java
@@ -0,0 +1,10 @@
+package io.quarkus.vertx.http.deployment;
+
+import io.quarkus.builder.item.MultiBuildItem;
+
+/**
+ * This is a marker that indicates that the body handler should be installed
+ * on all routes, as an extension requires the request to be fully buffered.
+ */
+public final class RequireBodyHandlerBuildItem extends MultiBuildItem {
+}
diff --git a/...tx-http/deployment/src/main/java/io/quarkus/vertx/http/deployment/VertxHttpProcessor.java b/...tx-http/deployment/src/main/java/io/quarkus/vertx/http/deployment/VertxHttpProcessor.java
@@ -35,8 +35,10 @@
 import io.quarkus.vertx.http.runtime.VertxHttpRecorder;
 import io.quarkus.vertx.http.runtime.cors.CORSRecorder;
 import io.quarkus.vertx.http.runtime.filters.Filter;
+import io.vertx.core.Handler;
 import io.vertx.core.impl.VertxImpl;
 import io.vertx.ext.web.Router;
+import io.vertx.ext.web.RoutingContext;
 
 class VertxHttpProcessor {
 
@@ -97,6 +99,12 @@ VertxWebRouterBuildItem initializeRouter(VertxHttpRecorder recorder,
         return new VertxWebRouterBuildItem(router);
     }
 
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    BodyHandlerBuildItem bodyHandler(VertxHttpRecorder recorder, HttpConfiguration httpConfiguration) {
+        return new BodyHandlerBuildItem(recorder.createBodyHandler(httpConfiguration));
+    }
+
     @BuildStep
     @Record(ExecutionTime.RUNTIME_INIT)
     ServiceStartBuildItem finalizeRouter(
@@ -106,7 +114,9 @@ ServiceStartBuildItem finalizeRouter(
             List<DefaultRouteBuildItem> defaultRoutes, List<FilterBuildItem> filters,
             VertxWebRouterBuildItem router, EventLoopCountBuildItem eventLoopCount,
             HttpBuildTimeConfig httpBuildTimeConfig, HttpConfiguration httpConfiguration,
-            BuildProducer<ReflectiveClassBuildItem> reflectiveClass, List<WebsocketSubProtocolsBuildItem> websocketSubProtocols)
+            BuildProducer<ReflectiveClassBuildItem> reflectiveClass, List<WebsocketSubProtocolsBuildItem> websocketSubProtocols,
+            List<RequireBodyHandlerBuildItem> requireBodyHandlerBuildItems,
+            BodyHandlerBuildItem bodyHandlerBuildItem)
             throws BuildException, IOException {
         Optional<DefaultRouteBuildItem> defaultRoute;
         if (defaultRoutes == null || defaultRoutes.isEmpty()) {
@@ -124,9 +134,13 @@ ServiceStartBuildItem finalizeRouter(
                 .filter(f -> f.getHandler() != null)
                 .map(FilterBuildItem::toFilter).collect(Collectors.toList());
 
+        //if the body handler is required then we know it is installed for all routes, so we don't need to register it here
+        Handler<RoutingContext> bodyHandler = requireBodyHandlerBuildItems.isEmpty() ? bodyHandlerBuildItem.getHandler() : null;
+
         recorder.finalizeRouter(beanContainer.getValue(),
                 defaultRoute.map(DefaultRouteBuildItem::getRoute).orElse(null),
-                listOfFilters, vertx.getVertx(), router.getRouter(), httpBuildTimeConfig.rootPath, launchMode.getLaunchMode());
+                listOfFilters, vertx.getVertx(), router.getRouter(), httpBuildTimeConfig.rootPath, launchMode.getLaunchMode(),
+                !requireBodyHandlerBuildItems.isEmpty(), bodyHandler);
 
         boolean startVirtual = requireVirtual.isPresent() || httpBuildTimeConfig.virtual;
         if (startVirtual) {

diff --git a/...ons/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/VertxHttpRecorder.java b/...ons/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/VertxHttpRecorder.java
@@ -35,6 +35,7 @@
 import io.quarkus.runtime.Timing;
 import io.quarkus.runtime.annotations.Recorder;
 import io.quarkus.runtime.configuration.ConfigInstantiator;
+import io.quarkus.runtime.configuration.MemorySize;
 import io.quarkus.vertx.core.runtime.VertxCoreRecorder;
 import io.quarkus.vertx.core.runtime.config.VertxConfiguration;
 import io.quarkus.vertx.http.runtime.filters.Filter;
@@ -61,6 +62,7 @@
 import io.vertx.ext.web.Route;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
+import io.vertx.ext.web.handler.BodyHandler;
 
 @Recorder
 public class VertxHttpRecorder {
@@ -169,7 +171,8 @@ public void startServer(RuntimeValue<Vertx> vertxRuntimeValue, ShutdownContext s
 
     public void finalizeRouter(BeanContainer container, Consumer<Route> defaultRouteHandler,
             List<Filter> filterList, RuntimeValue<Vertx> vertx,
-            RuntimeValue<Router> runtimeValue, String rootPath, LaunchMode launchMode) {
+            RuntimeValue<Router> runtimeValue, String rootPath, LaunchMode launchMode, boolean requireBodyHandler,
+            Handler<RoutingContext> bodyHandler) {
         // install the default route at the end
         Router router = runtimeValue.getValue();
 
@@ -200,6 +203,18 @@ public void finalizeRouter(BeanContainer container, Consumer<Route> defaultRoute
         container.instance(RouterProducer.class).initialize(resumingRouter);
         router.route().last().failureHandler(new QuarkusErrorHandler(launchMode.isDevOrTest()));
 
+        if (requireBodyHandler) {
+            //if this is set then everything needs the body handler installed
+            //TODO: config etc
+            router.route().order(Integer.MIN_VALUE).handler(new Handler<RoutingContext>() {
+                @Override
+                public void handle(RoutingContext routingContext) {
+                    routingContext.request().resume();
+                    bodyHandler.handle(routingContext);
+                }
+            });
+        }
+
         if (rootPath.equals("/")) {
             if (hotReplacementHandler != null) {
                 router.route().order(-1).handler(hotReplacementHandler);
@@ -626,4 +641,24 @@ public static Handler<HttpServerRequest> getRootHandler() {
         return ACTUAL_ROOT;
     }
 
+    public Handler<RoutingContext> createBodyHandler(HttpConfiguration httpConfiguration) {
+        BodyHandler bodyHandler = BodyHandler.create();
+        Optional<MemorySize> maxBodySize = httpConfiguration.limits.maxBodySize;
+        if (maxBodySize.isPresent()) {
+            bodyHandler.setBodyLimit(maxBodySize.get().asLongValue());
+        }
+        final BodyConfig bodyConfig = httpConfiguration.body;
+        bodyHandler.setHandleFileUploads(bodyConfig.handleFileUploads);
+        bodyHandler.setUploadsDirectory(bodyConfig.uploadsDirectory);
+        bodyHandler.setDeleteUploadedFilesOnEnd(bodyConfig.deleteUploadedFilesOnEnd);
+        bodyHandler.setMergeFormAttributes(bodyConfig.mergeFormAttributes);
+        bodyHandler.setPreallocateBodyBuffer(bodyConfig.preallocateBodyBuffer);
+        return new Handler<RoutingContext>() {
+            @Override
+            public void handle(RoutingContext event) {
+                event.request().resume();
+                bodyHandler.handle(event);
+            }
+        };
+    }
 }
diff --git a/...ions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/VertxInputStream.java b/...ions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/VertxInputStream.java
@@ -81,7 +81,7 @@ public int available() throws IOException {
         if (finished) {
             return -1;
         }
-        
+
         return exchange.readBytesAvailable();
     }
 
@@ -201,13 +201,13 @@ public int readBytesAvailable() {
             if (input1 != null) {
                 return input1.getByteBuf().readableBytes();
             }
-            
+
             String length = request.getHeader(HttpHeaders.CONTENT_LENGTH);
-            
+
             if (length == null) {
                 return 0;
             }
-            
+
             return Integer.parseInt(length);
         }
     }

diff --git a/...tx-web/deployment/src/main/java/io/quarkus/vertx/web/deployment/BodyHandlerBuildItem.java b/...tx-web/deployment/src/main/java/io/quarkus/vertx/web/deployment/BodyHandlerBuildItem.java
@@ -4,6 +4,10 @@
 import io.vertx.core.Handler;
 import io.vertx.ext.web.RoutingContext;
 
+/**
+ * use {@link io.quarkus.vertx.http.deployment.BodyHandlerBuildItem} instead
+ */
+@Deprecated
 public final class BodyHandlerBuildItem extends SimpleBuildItem {
     private final Handler<RoutingContext> handler;
 

diff --git a/...vertx-web/deployment/src/main/java/io/quarkus/vertx/web/deployment/VertxWebProcessor.java b/...vertx-web/deployment/src/main/java/io/quarkus/vertx/web/deployment/VertxWebProcessor.java
@@ -48,9 +48,9 @@
 import io.quarkus.gizmo.MethodDescriptor;
 import io.quarkus.gizmo.ResultHandle;
 import io.quarkus.vertx.http.deployment.FilterBuildItem;
+import io.quarkus.vertx.http.deployment.RequireBodyHandlerBuildItem;
 import io.quarkus.vertx.http.deployment.RouteBuildItem;
 import io.quarkus.vertx.http.runtime.HandlerType;
-import io.quarkus.vertx.http.runtime.HttpConfiguration;
 import io.quarkus.vertx.web.Route;
 import io.quarkus.vertx.web.RouteBase;
 import io.quarkus.vertx.web.RouteFilter;
@@ -149,9 +149,8 @@ void validateBeanDeployment(
     }
 
     @BuildStep
-    @Record(ExecutionTime.RUNTIME_INIT)
-    BodyHandlerBuildItem bodyHandler(VertxWebRecorder recorder, HttpConfiguration httpConfiguration) {
-        return new BodyHandlerBuildItem(recorder.createBodyHandler(httpConfiguration));
+    BodyHandlerBuildItem bodyHandler(io.quarkus.vertx.http.deployment.BodyHandlerBuildItem realOne) {
+        return new BodyHandlerBuildItem(realOne.getHandler());
     }
 
     @BuildStep
@@ -163,9 +162,10 @@ void addAdditionalRoutes(
             BuildProducer<GeneratedClassBuildItem> generatedClass,
             AnnotationProxyBuildItem annotationProxy,
             BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
-            BodyHandlerBuildItem bodyHandler,
+            io.quarkus.vertx.http.deployment.BodyHandlerBuildItem bodyHandler,
             BuildProducer<RouteBuildItem> routeProducer,
-            BuildProducer<FilterBuildItem> filterProducer) throws IOException {
+            BuildProducer<FilterBuildItem> filterProducer,
+            List<RequireBodyHandlerBuildItem> bodyHandlerRequired) throws IOException {
 
         ClassOutput classOutput = new GeneratedClassGizmoAdaptor(generatedClass, true);