Leverage platform http service in camel-knative #264

camel-k-main/camel-k-runtime-health/src/main/java/org/apache/camel/k/health/HealthContextCustomizer.java

@@ -24,7 +24,6 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Route;
-import io.vertx.ext.web.Router;
 import org.apache.camel.CamelContext;
 import org.apache.camel.Ordered;
 import org.apache.camel.health.HealthCheck;
@@ -35,7 +34,7 @@
 import org.apache.camel.impl.health.ContextHealthCheck;
 import org.apache.camel.impl.health.RoutesHealthCheckRepository;
 import org.apache.camel.k.ContextCustomizer;
-import org.apache.camel.k.http.PlatformHttpRouter;
+import org.apache.camel.k.http.PlatformHttp;
 
 public class HealthContextCustomizer implements ContextCustomizer {
     public static final String DEFAULT_PATH = "/health";
@@ -110,12 +109,17 @@ public void apply(CamelContext camelContext) {
         // add health route
         addRoute(
             camelContext,
-            PlatformHttpRouter.lookup(camelContext).get()
+            PlatformHttp.lookup(camelContext)
         );
     }
 
-    private Route addRoute(CamelContext camelContext, Router router) {
-        return router.route(HttpMethod.GET, path).handler(routingContext -> {
+    private Route addRoute(CamelContext camelContext, PlatformHttp platformHttp) {
+        Route route = platformHttp.router().route(HttpMethod.GET, path);
+
+        // add global handlers first i.e. body handler
+        platformHttp.handlers().forEach(route::handler);
+
+        route.handler(routingContext -> {
             int code = 200;
 
             Collection<HealthCheck.Result> results = HealthCheckHelper.invoke(
@@ -158,5 +162,7 @@ private Route addRoute(CamelContext camelContext, Router router) {
                 .setStatusCode(code)
                 .end(Json.encodePrettily(response));
         });
+
+        return route;
     }
 }

camel-k-main/pom.xml

@@ -31,7 +31,6 @@
     <modules>
         <module>camel-k-runtime-main</module>
         <module>camel-k-runtime-health</module>
-        <module>camel-k-runtime-http</module>
     </modules>
 
 </project>

camel-k-quarkus/camel-k-quarkus-knative/deployment/src/main/java/org/apache/camel/k/quarkus/knative/deployment/DeploymentProcessor.java

@@ -22,6 +22,8 @@
 import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
 import io.quarkus.vertx.core.deployment.CoreVertxBuildItem;
+import io.quarkus.vertx.http.deployment.BodyHandlerBuildItem;
+import io.quarkus.vertx.http.deployment.VertxWebRouterBuildItem;
 import org.apache.camel.component.knative.spi.KnativeEnvironment;
 import org.apache.camel.k.quarkus.knative.KnativeRecorder;
 import org.apache.camel.quarkus.core.deployment.CamelRuntimeBeanBuildItem;
@@ -41,11 +43,19 @@ void servicesFilters(BuildProducer<CamelServiceFilterBuildItem> serviceFilter) {
 
     @Record(ExecutionTime.RUNTIME_INIT)
     @BuildStep
-    CamelRuntimeBeanBuildItem knativeComponent(KnativeRecorder recorder, CoreVertxBuildItem vertx) {
+    CamelRuntimeBeanBuildItem knativeComponent(
+        KnativeRecorder recorder,
+        CoreVertxBuildItem vertx,
+        VertxWebRouterBuildItem router,
+        BodyHandlerBuildItem bodyHandlerBuildItem) {
+
         return new CamelRuntimeBeanBuildItem(
             "knative",
             "org.apache.camel.component.knative.KnativeComponent",
-            recorder.createKnativeComponent(vertx.getVertx())
+            recorder.createKnativeComponent(
+                vertx.getVertx(),
+                router.getRouter(),
+                bodyHandlerBuildItem.getHandler())
         );
     }
 

camel-k-quarkus/camel-k-quarkus-knative/runtime/src/main/java/org/apache/camel/k/quarkus/knative/KnativeRecorder.java

@@ -16,19 +16,32 @@
  */
 package org.apache.camel.k.quarkus.knative;
 
+import java.util.Collections;
 import java.util.function.Supplier;
 
 import io.quarkus.runtime.RuntimeValue;
 import io.quarkus.runtime.annotations.Recorder;
+import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
+import io.vertx.ext.web.Router;
+import io.vertx.ext.web.RoutingContext;
 import org.apache.camel.component.knative.KnativeComponent;
 import org.apache.camel.component.knative.http.KnativeHttpTransport;
+import org.apache.camel.k.http.PlatformHttp;
 
 @Recorder
 public class KnativeRecorder {
-    public RuntimeValue<KnativeComponent> createKnativeComponent(Supplier<Vertx> vertx) {
+    public RuntimeValue<KnativeComponent> createKnativeComponent(
+        Supplier<Vertx> vertx,
+        RuntimeValue<Router> router,
+        Handler<RoutingContext> bodyHandler) {
+
         KnativeHttpTransport transport = new KnativeHttpTransport();
-        transport.setVertx(vertx.get());
+        transport.setPlatformHttp(new PlatformHttp(
+            vertx.get(),
+            router.getValue(),
+            Collections.singletonList(bodyHandler)
+        ));
 
         KnativeComponent component = new KnativeComponent();
         component.setTransport(transport);

camel-k-main/camel-k-runtime-http/pom.xml → camel-k-runtime-http/pom.xml

@@ -20,9 +20,10 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <groupId>org.apache.camel.k</groupId>
-        <artifactId>camel-k-main</artifactId>
+        <artifactId>camel-k-runtime-parent</artifactId>
         <version>1.2.2-SNAPSHOT</version>
     </parent>
+
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>camel-k-runtime-http</artifactId>

camel-k-main/camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttpRouter.java → camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttp.java

@@ -16,35 +16,60 @@
  */
 package org.apache.camel.k.http;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
 import io.vertx.ext.web.Router;
+import io.vertx.ext.web.RoutingContext;
 import org.apache.camel.CamelContext;
 import org.apache.camel.component.platform.http.PlatformHttpConstants;
 import org.apache.camel.support.CamelContextHelper;
 
-public class PlatformHttpRouter {
+public class PlatformHttp {
     public static final String PLATFORM_HTTP_ROUTER_NAME = PlatformHttpConstants.PLATFORM_HTTP_COMPONENT_NAME + "-router";
 
+    private final Vertx vertx;
     private final Router router;
+    private final List<Handler<RoutingContext>> handlers;
+
+    public PlatformHttp(Vertx vertx, Router router) {
+        this.vertx = vertx;
+        this.router = router;
+        this.handlers = Collections.emptyList();
+    }
 
-    public PlatformHttpRouter(Router router) {
+    public PlatformHttp(Vertx vertx, Router router, List<Handler<RoutingContext>> handlers) {
+        this.vertx = vertx;
         this.router = router;
+        this.handlers = Collections.unmodifiableList(new ArrayList<>(handlers));
     }
 
-    public Router get() {
+    public Vertx vertx() {
+        return vertx;
+    }
+
+    public Router router() {
         return router;
     }
 
+    public List<Handler<RoutingContext>> handlers() {
+        return handlers;
+    }
+
     // **********************
     //
     // Helpers
     //
     // **********************
 
-    public static PlatformHttpRouter lookup(CamelContext camelContext) {
+    public static PlatformHttp lookup(CamelContext camelContext) {
         return CamelContextHelper.mandatoryLookup(
             camelContext,
-            PlatformHttpRouter.PLATFORM_HTTP_ROUTER_NAME,
-            PlatformHttpRouter.class
+            PlatformHttp.PLATFORM_HTTP_ROUTER_NAME,
+            PlatformHttp.class
         );
     }
 }

camel-k-main/camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttpServer.java → camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttpServer.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.k.http;
 
+import java.util.Collections;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CountDownLatch;
@@ -33,7 +34,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-final class PlatformHttpServer extends ServiceSupport {
+public final class PlatformHttpServer extends ServiceSupport {
     private static final Logger LOGGER = LoggerFactory.getLogger(PlatformHttpServer.class);
 
     private final CamelContext context;
@@ -70,16 +71,12 @@ private CompletionStage<Void> startAsync() {
         final Router router = Router.router(vertx);
         final Router subRouter = Router.router(vertx);
 
-        router.route()
-            .order(Integer.MIN_VALUE)
-            .handler(ctx -> {
-                ctx.request().resume();
-                createBodyHandler().handle(ctx);
-            });
-
         router.mountSubRouter(configuration.getPath(), subRouter);
 
-        context.getRegistry().bind(PlatformHttpRouter.PLATFORM_HTTP_ROUTER_NAME, new PlatformHttpRouter(subRouter));
+        context.getRegistry().bind(
+            PlatformHttp.PLATFORM_HTTP_ROUTER_NAME,
+            new PlatformHttp(vertx, subRouter, Collections.singletonList(createBodyHandler()))
+        );
 
         //HttpServerOptions options = new HttpServerOptions();
         if (configuration.getSslContextParameters() != null) {

camel-k-main/camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttpServiceConfiguration.java → camel-k-runtime-http/src/main/java/org/apache/camel/k/http/PlatformHttpServiceConfiguration.java

@@ -30,7 +30,7 @@ public class PlatformHttpServiceConfiguration {
     private String path = DEFAULT_PATH;
     private BigInteger maxBodySize;
 
-    private BodyHandler bodyHandler = new BodyHandler();
+    private BodyHandlerConfiguration bodyHandlerConfiguration = new BodyHandlerConfiguration();
     private SSLContextParameters sslContextParameters;
 
     public String getBindHost() {
@@ -65,12 +65,12 @@ public void setMaxBodySize(BigInteger maxBodySize) {
         this.maxBodySize = maxBodySize;
     }
 
-    public BodyHandler getBodyHandler() {
-        return bodyHandler;
+    public BodyHandlerConfiguration getBodyHandler() {
+        return bodyHandlerConfiguration;
     }
 
-    public void setBodyHandler(BodyHandler bodyHandler) {
-        this.bodyHandler = bodyHandler;
+    public void setBodyHandler(BodyHandlerConfiguration bodyHandler) {
+        this.bodyHandlerConfiguration = bodyHandler;
     }
 
     public SSLContextParameters getSslContextParameters() {
@@ -81,7 +81,7 @@ public void setSslContextParameters(SSLContextParameters sslContextParameters) {
         this.sslContextParameters = sslContextParameters;
     }
 
-    public static class BodyHandler {
+    public static class BodyHandlerConfiguration {
         private boolean handleFileUploads = true;
         private String uploadsDirectory = "file-uploads";
         private boolean mergeFormAttributes = true;

camel-k-main/camel-k-runtime-http/src/main/java/org/apache/camel/k/http/engine/RuntimePlatformHttpConsumer.java → camel-k-runtime-http/src/main/java/org/apache/camel/k/http/engine/RuntimePlatformHttpConsumer.java

@@ -45,7 +45,7 @@
 import org.apache.camel.TypeConverter;
 import org.apache.camel.component.platform.http.PlatformHttpEndpoint;
 import org.apache.camel.component.platform.http.spi.Method;
-import org.apache.camel.k.http.PlatformHttpRouter;
+import org.apache.camel.k.http.PlatformHttp;
 import org.apache.camel.spi.HeaderFilterStrategy;
 import org.apache.camel.support.DefaultConsumer;
 import org.apache.camel.support.DefaultMessage;
@@ -75,10 +75,12 @@ protected void doStart() throws Exception {
         super.doStart();
 
         final PlatformHttpEndpoint endpoint = getEndpoint();
-        final PlatformHttpRouter router = PlatformHttpRouter.lookup(endpoint.getCamelContext());
+        final PlatformHttp router = PlatformHttp.lookup(endpoint.getCamelContext());
         final String path = endpoint.getPath();
         final String vertxPathParamPath = PATH_PARAMETER_PATTERN.matcher(path).replaceAll(":$1");
-        final Route newRoute = router.get().route(vertxPathParamPath);
+        final Route newRoute = router.router().route(vertxPathParamPath);
+
+        router.handlers().forEach(newRoute::handler);
 
         final Set<Method> methods = Method.parseList(endpoint.getHttpMethodRestrict());
         if (!methods.equals(Method.getAll())) {

camel-k-main/camel-k-runtime-http/src/test/java/org/apache/camel/k/http/PlatformHttpServiceCustomizerTest.java → camel-k-runtime-http/src/test/java/org/apache/camel/k/http/PlatformHttpServiceCustomizerTest.java

@@ -50,7 +50,7 @@ public void testPlatformHttpServiceCustomizer(String path) throws Exception {
 
         httpService.apply(runtime.getCamelContext());
 
-        PlatformHttpRouter.lookup(runtime.getCamelContext()).get().route(HttpMethod.GET, "/my/path")
+        PlatformHttp.lookup(runtime.getCamelContext()).router().route(HttpMethod.GET, "/my/path")
             .handler(routingContext -> {
                 JsonObject response = new JsonObject();
                 response.put("status", "UP");

camel-k-runtime-knative/src/test/java/org/apache/camel/k/loader/knative/KnativeSourceRoutesLoaderTest.java

@@ -35,6 +35,7 @@
 import org.apache.camel.k.Source;
 import org.apache.camel.k.SourceLoader;
 import org.apache.camel.k.Sources;
+import org.apache.camel.k.http.PlatformHttpServiceContextCustomizer;
 import org.apache.camel.k.listener.RoutesConfigurer;
 import org.apache.camel.k.loader.java.JavaSourceLoader;
 import org.apache.camel.model.ModelCamelContext;
@@ -69,6 +70,7 @@ public void testWrapLoader(String uri) throws Exception {
         LOGGER.info("uri: {}", uri);
 
         final int port = AvailablePortFinder.getNextAvailable();
+        final int platformHttpPort = AvailablePortFinder.getNextAvailable();
         final String data = UUID.randomUUID().toString();
 
         KnativeComponent component = new KnativeComponent();
@@ -173,10 +175,16 @@ public void testWrapLoaderWithBeanRegistration() throws Exception {
     static class TestRuntime implements Runtime {
         private final CamelContext camelContext;
         private final List<RoutesBuilder> builders;
+        private final int platformHttpPort;
 
         public TestRuntime() {
             this.camelContext = new DefaultCamelContext();
             this.builders = new ArrayList<>();
+            this.platformHttpPort = AvailablePortFinder.getNextAvailable();
+
+            PlatformHttpServiceContextCustomizer httpService = new PlatformHttpServiceContextCustomizer();
+            httpService.setBindPort(platformHttpPort);
+            httpService.apply(this.camelContext);
         }
 
         @Override

camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java

@@ -124,6 +124,17 @@ public static KnativeServiceDefinition endpoint(Knative.EndpointKind endpointKin
         );
     }
 
+    public static KnativeServiceDefinition sourceEndpoint(String name, Map<String, String> metadata) {
+        return entry(
+            Knative.EndpointKind.source,
+            Knative.Type.endpoint,
+            name,
+            null,
+            -1,
+            metadata
+        );
+    }
+
     public static KnativeServiceDefinition channel(Knative.EndpointKind endpointKind, String name, String host, int port) {
         return entry(
             endpointKind,
@@ -157,6 +168,28 @@ public static KnativeServiceDefinition event(Knative.EndpointKind endpointKind,
         );
     }
 
+    public static KnativeServiceDefinition sourceEvent(String name) {
+        return entry(
+            Knative.EndpointKind.source,
+            Knative.Type.event,
+            name,
+            null,
+            -1,
+            Collections.emptyMap()
+        );
+    }
+
+    public static KnativeServiceDefinition sourceEvent(String name, Map<String, String> metadata) {
+        return entry(
+            Knative.EndpointKind.source,
+            Knative.Type.event,
+            name,
+            null,
+            -1,
+            metadata
+        );
+    }
+
     public static KnativeServiceDefinition event(Knative.EndpointKind endpointKind, String name, String host, int port, Map<String, String> metadata) {
         return entry(
             endpointKind,
@@ -198,8 +231,8 @@ public static final class KnativeServiceDefinition extends DefaultServiceDefinit
         public KnativeServiceDefinition(
             @JsonProperty(value = "type", required = true) Knative.Type type,
             @JsonProperty(value = "name", required = true) String name,
-            @JsonProperty(value = "host", required = true) String host,
-            @JsonProperty(value = "port", required = true) int port,
+            @JsonProperty(value = "host", required = false) String host,
+            @JsonProperty(value = "port", required = false) int port,
             @JsonProperty(value = "metadata", required = false) Map<String, String> metadata) {
 
             super(