[camel-k] branch master updated: cloudevents & knative

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

[camel-k] branch master updated: cloudevents & knative

nferraro
This is an automated email from the ASF dual-hosted git repository.

nferraro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k.git


The following commit(s) were added to refs/heads/master by this push:
     new 5fce954  cloudevents & knative
5fce954 is described below

commit 5fce9548f41f2693b89f9d1d886fecc0ee85ad87
Author: lburgazzoli <[hidden email]>
AuthorDate: Tue Oct 16 19:44:52 2018 +0200

    cloudevents & knative
---
 runtime/camel-knative/.gitignore                   |  10 +
 runtime/camel-knative/pom.xml                      | 130 ++++++++
 .../apache/camel/component/knative/Knative.java    |  44 +++
 .../camel/component/knative/KnativeComponent.java  | 111 +++++++
 .../component/knative/KnativeConfiguration.java    |  61 ++++
 .../camel/component/knative/KnativeEndpoint.java   | 233 +++++++++++++++
 .../component/knative/KnativeEnvironment.java      | 173 +++++++++++
 .../camel/component/knative/KnativeProducer.java   |  79 +++++
 .../camel/component/knative/KnativeSupport.java    |  46 +++
 .../services/org/apache/camel/component/knative    |  18 ++
 .../component/knative/KnativeComponentTest.java    | 327 +++++++++++++++++++++
 .../src/test/resources/environment.json            |  26 ++
 .../src/test/resources/log4j2.properties           |   7 +
 runtime/pom.xml                                    |   5 +
 14 files changed, 1270 insertions(+)

diff --git a/runtime/camel-knative/.gitignore b/runtime/camel-knative/.gitignore
new file mode 100644
index 0000000..ed92983
--- /dev/null
+++ b/runtime/camel-knative/.gitignore
@@ -0,0 +1,10 @@
+target
+
+*.iml
+
+.idea
+.project
+.metadata
+.settings
+.factorypath
+.classpath
diff --git a/runtime/camel-knative/pom.xml b/runtime/camel-knative/pom.xml
new file mode 100644
index 0000000..53088a0
--- /dev/null
+++ b/runtime/camel-knative/pom.xml
@@ -0,0 +1,130 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+         http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<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-runtime-parent</artifactId>
+        <version>0.0.5-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>camel-knative</artifactId>
+
+    <dependencies>
+
+        <!-- ****************************** -->
+        <!--                                -->
+        <!-- RUNTIME                        -->
+        <!--                                -->
+        <!-- ****************************** -->
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-undertow</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jdk8</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-collections4</artifactId>
+            <version>${commons-collections4.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons-lang.version}</version>
+        </dependency>
+
+        <!-- ****************************** -->
+        <!--                                -->
+        <!-- TESTS                          -->
+        <!--                                -->
+        <!-- ****************************** -->
+
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-test</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>junit</groupId>
+                    <artifactId>junit</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>${junit-jupiter.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <version>${junit-jupiter.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+            <version>${log4j2.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <version>${log4j2.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/Knative.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/Knative.java
new file mode 100644
index 0000000..733555f
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/Knative.java
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+
+public final class Knative {
+    public static final ObjectMapper MAPPER = new ObjectMapper().registerModule(new Jdk8Module());
+
+    public static final String HTTP_COMPONENT = "undertow";
+    public static final String KNATIVE_PROTOCOL = "knative.protocol";
+    public static final String KNATIVE_TYPE = "knative.type";
+    public static final String KNATIVE_EVENT_TYPE = "knative.event.type";
+    public static final String CONTENT_TYPE = "content.type";
+    public static final String MIME_STRUCTURED_CONTENT_MODE = "application/cloudevents+json";
+
+    private Knative() {
+    }
+
+    public enum Type {
+        endpoint,
+        channel
+    }
+
+    public enum Protocol {
+        http,
+        https
+    }
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeComponent.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeComponent.java
new file mode 100644
index 0000000..0c88d0c
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeComponent.java
@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.StringHelper;
+
+public class KnativeComponent extends DefaultComponent {
+    private final KnativeConfiguration configuration;
+    private String environmentPath;
+
+    public KnativeComponent() {
+        this(null);
+    }
+
+    public KnativeComponent(CamelContext context) {
+        super(context);
+
+        this.configuration = new KnativeConfiguration();
+    }
+
+    // ************************
+    //
+    // Properties
+    //
+    // ************************
+
+    public String getEnvironmentPath() {
+        return environmentPath;
+    }
+
+    /**
+     * The path ot the environment definition
+     */
+    public void setEnvironmentPath(String environmentPath) {
+        this.environmentPath = environmentPath;
+    }
+
+    public KnativeConfiguration getConfiguration() {
+        return configuration;
+    }
+
+    public KnativeEnvironment getEnvironment() {
+        return configuration.getEnvironment();
+    }
+
+    /**
+     * The environment
+     */
+    public void setEnvironment(KnativeEnvironment environment) {
+        configuration.setEnvironment(environment);
+    }
+
+    // ************************
+    //
+    //
+    //
+    // ************************
+
+    @Override
+    protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+        final String type = StringHelper.before(remaining, "/");
+        final String target = StringHelper.after(remaining, "/");
+        final KnativeConfiguration conf = getKnativeConfiguration();
+
+        // set properties from the endpoint uri
+        IntrospectionSupport.setProperties(getCamelContext().getTypeConverter(), conf, parameters);
+
+        return new KnativeEndpoint(uri, this, Knative.Type.valueOf(type), target, conf);
+    }
+
+    // ************************
+    //
+    // Helpers
+    //
+    // ************************
+
+    private KnativeConfiguration getKnativeConfiguration() throws Exception {
+        KnativeConfiguration conf = configuration.copy();
+
+        if (conf.getEnvironment() == null) {
+            ObjectHelper.notNull(environmentPath, "Environment Path");
+
+            conf.setEnvironment(
+                KnativeEnvironment.mandatoryLoadFromResource(getCamelContext(), this.environmentPath)
+            );
+        }
+
+        return conf;
+    }
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeConfiguration.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeConfiguration.java
new file mode 100644
index 0000000..cc3fa65
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeConfiguration.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+
+public class KnativeConfiguration implements Cloneable {
+    @UriParam
+    @Metadata(required = "true")
+    private KnativeEnvironment environment;
+
+    public KnativeConfiguration() {
+    }
+
+    // ************************
+    //
+    // Properties
+    //
+    // ************************
+
+    public KnativeEnvironment getEnvironment() {
+        return environment;
+    }
+
+    /**
+     * The environment
+     */
+    public void setEnvironment(KnativeEnvironment environment) {
+        this.environment = environment;
+    }
+
+    // ************************
+    //
+    // Cloneable
+    //
+    // ************************
+
+    public KnativeConfiguration copy() {
+        try {
+            return (KnativeConfiguration)super.clone();
+        } catch (CloneNotSupportedException e) {
+            throw new RuntimeCamelException(e);
+        }
+    }
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
new file mode 100644
index 0000000..986ebeb
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
@@ -0,0 +1,233 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.io.InputStream;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.DelegateEndpoint;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.processor.Pipeline;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.camel.util.StringHelper;
+import org.apache.commons.lang3.StringUtils;
+
+import static org.apache.camel.util.ObjectHelper.ifNotEmpty;
+
+
+@UriEndpoint(
+    firstVersion = "3.0.0",
+    scheme = "knative",
+    syntax = "knative:type/target",
+    title = "Knative",
+    producerOnly = true,
+    label = "cloud,eventing")
+public class KnativeEndpoint extends DefaultEndpoint implements DelegateEndpoint {
+    @UriPath(description = "The Knative type")
+    @Metadata(required = "true")
+    private final Knative.Type type;
+
+    @UriPath(description = "The Knative name")
+    @Metadata(required = "true")
+    private final String name;
+
+    @UriParam
+    private final KnativeConfiguration configuration;
+
+    private final KnativeEnvironment environment;
+    private final KnativeEnvironment.KnativeServiceDefinition service;
+    private final Endpoint endpoint;
+
+    public KnativeEndpoint(String uri, KnativeComponent component, Knative.Type targetType, String remaining, KnativeConfiguration configuration) {
+        super(uri, component);
+
+        this.type = targetType;
+        this.name = remaining.indexOf('/') != -1 ? StringHelper.before(remaining, "/") : remaining;
+        this.configuration = configuration;
+        this.environment =  this.configuration.getEnvironment();
+        this.service = this.environment.mandatoryLookupService(targetType, remaining);
+
+        switch (service.getProtocol()) {
+        case http:
+            this.endpoint = http(component.getCamelContext(), service);
+            break;
+        case https:
+            this.endpoint = http(component.getCamelContext(), service);
+            break;
+        default:
+            throw new IllegalArgumentException("unsupported protocol: " + this.service.getProtocol());
+        }
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+        ServiceHelper.startService(endpoint);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        ServiceHelper.stopService(endpoint);
+        super.doStop();
+    }
+
+    @Override
+    public Producer createProducer() throws Exception {
+        return new KnativeProducer(
+            this,
+            exchange -> {
+                final String eventType = service.getMetadata().get(Knative.KNATIVE_EVENT_TYPE);
+                final String contentType = service.getMetadata().get(Knative.CONTENT_TYPE);
+                final ZonedDateTime created = exchange.getCreated().toInstant().atZone(ZoneId.systemDefault());
+                final String eventTime = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(created);
+                final Map<String, Object> headers = exchange.getIn().getHeaders();
+
+                headers.putIfAbsent("CE-CloudEventsVersion", "0.1");
+                headers.putIfAbsent("CE-EventType", eventType);
+                headers.putIfAbsent("CE-EventID", exchange.getExchangeId());
+                headers.putIfAbsent("CE-EventTime", eventTime);
+                headers.putIfAbsent("CE-Source", getEndpointUri());
+                headers.putIfAbsent(Exchange.CONTENT_TYPE, contentType);
+            },
+            endpoint.createProducer()
+        );
+    }
+
+    @Override
+    public Consumer createConsumer(Processor processor) throws Exception {
+        final Processor pipeline = Pipeline.newInstance(
+            getCamelContext(),
+            exchange -> {
+                if (!KnativeSupport.hasStructuredContent(exchange)) {
+                    //
+                    // The event is not in the form of Structured Content Mode
+                    // then leave it as it is.
+                    //
+                    // Note that this is true for http binding only.
+                    //
+                    // More info:
+                    //
+                    //   https://github.com/cloudevents/spec/blob/master/http-transport-binding.md#32-structured-content-mode
+                    //
+                    return;
+                }
+
+                try (InputStream is = exchange.getIn().getBody(InputStream.class)) {
+                    final Message message = exchange.getIn();
+                    final Map<String, Object> ce = Knative.MAPPER.readValue(is, Map.class);
+
+                    ifNotEmpty(ce.remove("contentType"), val -> message.setHeader(Exchange.CONTENT_TYPE, val));
+                    ifNotEmpty(ce.remove("data"), val -> message.setBody(val));
+
+                    //
+                    // Map extensions to standard camel headers
+                    //
+                    ifNotEmpty(ce.remove("extensions"), val -> {
+                        if (val instanceof Map) {
+                            ((Map<String, Object>) val).forEach(message::setHeader);
+                        }
+                    });
+
+                    ce.forEach((key, val) -> {
+                        message.setHeader("CE-" + StringUtils.capitalize(key), val);
+                    });
+                }
+            },
+            processor
+        );
+
+        final Consumer consumer = endpoint.createConsumer(pipeline);
+
+        configureConsumer(consumer);
+
+        return consumer;
+    }
+
+    @Override
+    public boolean isSingleton() {
+        return true;
+    }
+
+    @Override
+    public Endpoint getEndpoint() {
+        return this.endpoint;
+    }
+
+    public Knative.Type getType() {
+        return type;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public KnativeEnvironment.KnativeServiceDefinition getService() {
+        return service;
+    }
+
+    // *****************************
+    //
+    // Helpers
+    //
+    // *****************************
+
+    private static Endpoint http(CamelContext context, ServiceDefinition definition) {
+        try {
+            final String host = definition.getHost();
+            final int port = definition.getPort();
+            final String scheme = Knative.HTTP_COMPONENT;
+            final String protocol = definition.getMetadata().get(Knative.KNATIVE_PROTOCOL);
+
+            ObjectHelper.notNull(host, ServiceDefinition.SERVICE_META_HOST);
+            ObjectHelper.notNull(protocol, Knative.KNATIVE_PROTOCOL);
+
+            String uri = String.format("%s:%s://%s", scheme, protocol, host);
+            if (port != -1) {
+                uri = uri + ":" + port;
+            }
+
+            String path = definition.getMetadata().get(ServiceDefinition.SERVICE_META_PATH);
+            if (path != null) {
+                if (!path.startsWith("/")) {
+                    uri += "/";
+                }
+
+                uri += path;
+            }
+
+            return context.getEndpoint(uri);
+        } catch (Exception e) {
+            throw ObjectHelper.wrapRuntimeCamelException(e);
+        }
+    }
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEnvironment.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEnvironment.java
new file mode 100644
index 0000000..254ef8d
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEnvironment.java
@@ -0,0 +1,173 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.camel.CamelContext;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.impl.cloud.DefaultServiceDefinition;
+import org.apache.camel.util.CollectionHelper;
+import org.apache.camel.util.ResourceHelper;
+import org.apache.camel.util.StringHelper;
+
+/*
+ * Assuming it is loaded from a json for now
+ */
+public class KnativeEnvironment {
+    private final List<KnativeServiceDefinition> services;
+
+    @JsonCreator
+    public KnativeEnvironment(
+        @JsonProperty(value = "services", required = true) List<KnativeServiceDefinition> services) {
+
+        this.services = new ArrayList<>(services);
+    }
+
+    public Stream<KnativeServiceDefinition> stream() {
+        return services.stream();
+    }
+
+    public Optional<KnativeServiceDefinition> lookupService(Knative.Type type, String name) {
+        final String contextPath = StringHelper.after(name, "/");
+        final String serviceName = (contextPath == null) ? name : StringHelper.before(name, "/");
+
+        return services.stream()
+            .filter(definition -> {
+                return Objects.equals(type.name(), definition.getMetadata().get(Knative.KNATIVE_TYPE))
+                    && Objects.equals(serviceName, definition.getName());
+            })
+            .map(definition -> {
+                //
+                // The context path set on the endpoint  overrides the one
+                // eventually provided by the service definition.
+                //
+                if (contextPath != null) {
+                    return new KnativeServiceDefinition(
+                        definition.getType(),
+                        definition.getProtocol(),
+                        definition.getName(),
+                        definition.getHost(),
+                        definition.getPort(),
+                        KnativeSupport.mergeMaps(
+                            definition.getMetadata(),
+                            Collections.singletonMap(ServiceDefinition.SERVICE_META_PATH, "/" + contextPath)
+                        )
+                    );
+                }
+
+                return definition;
+            })
+            .findFirst();
+    }
+
+    public KnativeServiceDefinition mandatoryLookupService(Knative.Type type, String name) {
+        return lookupService(type, name).orElseThrow(
+            () -> new IllegalArgumentException("Unable to find the service \"" + name + "\" with type \"" + type + "\"")
+        );
+    }
+
+    // ************************
+    //
+    // Helpers
+    //
+    // ************************
+
+    public static KnativeEnvironment mandatoryLoadFromResource(CamelContext context, String path) throws Exception {
+        try (InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(context, path)) {
+
+            //
+            // read the knative environment from a file formatted as json, i.e. :
+            //
+            // {
+            //     "services": [
+            //         {
+            //              "type": "channel|endpoint",
+            //              "protocol": "http",
+            //              "name": "",
+            //              "host": "",
+            //              "port": "",
+            //              "metadata": {
+            //                  "service.path": "",
+            //                  "knative.event.type": ""
+            //              }
+            //         },
+            //     ]
+            // }
+            //
+            //
+            return Knative.MAPPER.readValue(is, KnativeEnvironment.class);
+        }
+    }
+
+    // ************************
+    //
+    // Types
+    //
+    // ************************
+
+    public final static class KnativeServiceDefinition extends DefaultServiceDefinition {
+        @JsonCreator
+        public KnativeServiceDefinition(
+            @JsonProperty(value = "type", required = true) Knative.Type type,
+            @JsonProperty(value = "protocol", required = true) Knative.Protocol protocol,
+            @JsonProperty(value = "name", required = true) String name,
+            @JsonProperty(value = "host", required = true) String host,
+            @JsonProperty(value = "port", required = true) int port,
+            @JsonProperty(value = "metadata", required = false) Map<String, String> metadata) {
+
+            super(
+                UUID.randomUUID().toString(),
+                name,
+                host,
+                port,
+                KnativeSupport.mergeMaps(
+                    metadata,
+                    CollectionHelper.mapOf(
+                        Knative.KNATIVE_TYPE, type.name(),
+                        Knative.KNATIVE_PROTOCOL, protocol.name())
+                )
+            );
+        }
+
+        public Knative.Type getType() {
+            return Knative.Type.valueOf(getMetadata().get(Knative.KNATIVE_TYPE));
+        }
+
+        public Knative.Protocol getProtocol() {
+            return Knative.Protocol.valueOf(getMetadata().get(Knative.KNATIVE_PROTOCOL));
+        }
+
+        public String getPath() {
+            return getMetadata().get(ServiceDefinition.SERVICE_META_PATH);
+        }
+
+        public String getEventType() {
+            return getMetadata().get(Knative.KNATIVE_EVENT_TYPE);
+        }
+    }
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeProducer.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeProducer.java
new file mode 100644
index 0000000..740455f
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeProducer.java
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.processor.Pipeline;
+import org.apache.camel.util.AsyncProcessorConverterHelper;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.commons.collections4.CollectionUtils;
+
+public class KnativeProducer extends DefaultAsyncProducer {
+    final AsyncProcessor processor;
+
+    public KnativeProducer(Endpoint endpoint, Processor processor, Processor... processors) {
+        super(endpoint);
+
+        List<Processor> elements = new ArrayList<>(1 + processors.length);
+
+        CollectionUtils.addAll(elements, processor);
+        CollectionUtils.addAll(elements, processors);
+
+        Processor pipeline = Pipeline.newInstance(endpoint.getCamelContext(), elements);
+
+        this.processor = AsyncProcessorConverterHelper.convert(pipeline);
+    }
+
+    @Override
+    public boolean process(Exchange exchange, AsyncCallback callback) {
+        return processor.process(exchange, callback);
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        ServiceHelper.startServices(processor);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        ServiceHelper.stopServices(processor);
+    }
+
+    @Override
+    protected void doSuspend() throws Exception {
+        ServiceHelper.suspendService(processor);
+    }
+
+    @Override
+    protected void doResume() throws Exception {
+        ServiceHelper.resumeService(processor);
+    }
+
+    @Override
+    protected void doShutdown() throws Exception {
+        ServiceHelper.stopAndShutdownServices(processor);
+    }
+
+}
diff --git a/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeSupport.java b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeSupport.java
new file mode 100644
index 0000000..9c6c049
--- /dev/null
+++ b/runtime/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeSupport.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.camel.Exchange;
+
+public final class KnativeSupport {
+    private KnativeSupport() {
+    }
+
+    public static boolean hasStructuredContent(Exchange exchange) {
+        return Objects.equals(exchange.getIn().getHeader(Exchange.CONTENT_TYPE), Knative.MIME_STRUCTURED_CONTENT_MODE);
+    }
+
+    public static <K, V> Map<K, V> mergeMaps(Map<K, V> map, Map<K, V>... maps) {
+        Map<K, V> answer = new HashMap<>();
+
+        if (map != null) {
+            answer.putAll(map);
+        }
+
+        for (Map<K, V> m : maps) {
+            answer.putAll(m);
+        }
+
+        return answer;
+    }
+}
diff --git a/runtime/camel-knative/src/main/resources/META-INF/services/org/apache/camel/component/knative b/runtime/camel-knative/src/main/resources/META-INF/services/org/apache/camel/component/knative
new file mode 100644
index 0000000..3d68812
--- /dev/null
+++ b/runtime/camel-knative/src/main/resources/META-INF/services/org/apache/camel/component/knative
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+class=org.apache.camel.component.knative.KnativeComponent
\ No newline at end of file
diff --git a/runtime/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java b/runtime/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java
new file mode 100644
index 0000000..ecc265a
--- /dev/null
+++ b/runtime/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java
@@ -0,0 +1,327 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.knative;
+
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.undertow.UndertowEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.component.knative.KnativeEnvironment.mandatoryLoadFromResource;
+import static org.apache.camel.util.CollectionHelper.mapOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class KnativeComponentTest {
+
+    private CamelContext context;
+
+    // **************************
+    //
+    // Setup
+    //
+    // **************************
+
+    @BeforeEach
+    public void before() {
+        this.context = new DefaultCamelContext();
+    }
+
+    @AfterEach
+    public void after() throws Exception {
+        if (this.context != null) {
+            this.context.stop();
+        }
+    }
+
+    // **************************
+    //
+    // Tests
+    //
+    // **************************
+
+    @Test
+    void testLoadEnvironment() throws Exception {
+        KnativeEnvironment env = mandatoryLoadFromResource(context, "classpath:/environment.json");
+
+        assertThat(env.stream()).hasSize(2);
+        assertThat(env.stream()).anyMatch(s -> s.getType() == Knative.Type.channel);
+        assertThat(env.stream()).anyMatch(s -> s.getType() == Knative.Type.endpoint);
+
+        assertThat(env.lookupService(Knative.Type.channel, "c1")).isPresent();
+        assertThat(env.lookupService(Knative.Type.channel, "e1")).isNotPresent();
+        assertThat(env.lookupService(Knative.Type.endpoint, "e1")).isPresent();
+        assertThat(env.lookupService(Knative.Type.endpoint, "c1")).isNotPresent();
+
+        assertThatThrownBy(() -> env.mandatoryLookupService(Knative.Type.endpoint, "unknown"))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Unable to find the service \"unknown\" with type \"endpoint\"");
+
+        assertThatThrownBy(() -> env.mandatoryLookupService(Knative.Type.channel, "unknown"))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Unable to find the service \"unknown\" with type \"channel\"");
+    }
+
+    @Test
+    void testCreateComponent() throws Exception {
+        context.start();
+
+        assertThat(context.getComponent("knative")).isNotNull();
+        assertThat(context.getComponent("knative")).isInstanceOf(KnativeComponent.class);
+    }
+
+    @Test
+    void testCreateEndpoint() throws Exception {
+        KnativeEnvironment env = new KnativeEnvironment(Arrays.asList(
+            new KnativeEnvironment.KnativeServiceDefinition(
+                Knative.Type.endpoint,
+                Knative.Protocol.http,
+                "myEndpoint",
+                "my-node",
+                9001,
+                mapOf(ServiceDefinition.SERVICE_META_PATH, "/a/path"))
+        ));
+
+        KnativeComponent component = context.getComponent("knative", KnativeComponent.class);
+        component.setEnvironment(env);
+
+        context.start();
+
+        //
+        // Endpoint with context path derived from service definition
+        //
+
+        KnativeEndpoint e1 = context.getEndpoint("knative:endpoint/myEndpoint", KnativeEndpoint.class);
+
+        assertThat(e1).isNotNull();
+        assertThat(e1.getService()).isNotNull();
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("name", "myEndpoint");
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("host", "my-node");
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("port", 9001);
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("type", Knative.Type.endpoint);
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("protocol", Knative.Protocol.http);
+        assertThat(e1.getService()).hasFieldOrPropertyWithValue("path", "/a/path");
+        assertThat(e1.getEndpoint()).isInstanceOf(UndertowEndpoint.class);
+        assertThat(e1.getEndpoint()).hasFieldOrPropertyWithValue("endpointUri", "http://my-node:9001/a/path");
+
+        //
+        // Endpoint with context path overridden by endpoint uri
+        //
+
+        KnativeEndpoint e2 = context.getEndpoint("knative:endpoint/myEndpoint/another/path", KnativeEndpoint.class);
+
+        assertThat(e2).isNotNull();
+        assertThat(e2.getService()).isNotNull();
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("name", "myEndpoint");
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("host", "my-node");
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("port", 9001);
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("type", Knative.Type.endpoint);
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("protocol", Knative.Protocol.http);
+        assertThat(e2.getService()).hasFieldOrPropertyWithValue("path", "/another/path");
+        assertThat(e2.getEndpoint()).isInstanceOf(UndertowEndpoint.class);
+        assertThat(e2.getEndpoint()).hasFieldOrPropertyWithValue("endpointUri", "http://my-node:9001/another/path");
+    }
+
+    @Test
+    void testInvokeEndpoint() throws Exception {
+        final int port = AvailablePortFinder.getNextAvailable();
+
+        KnativeEnvironment env = new KnativeEnvironment(Arrays.asList(
+            new KnativeEnvironment.KnativeServiceDefinition(
+                Knative.Type.endpoint,
+                Knative.Protocol.http,
+                "myEndpoint",
+                "localhost",
+                port,
+                mapOf(
+                    ServiceDefinition.SERVICE_META_PATH, "/a/path",
+                    Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+                    Knative.CONTENT_TYPE, "text/plain"
+                ))
+        ));
+
+        KnativeComponent component = context.getComponent("knative", KnativeComponent.class);
+        component.setEnvironment(env);
+
+        context.addRoutes(new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:source")
+                    .to("knative:endpoint/myEndpoint");
+
+                fromF("undertow:<a href="http://localhost:%d/a/path">http://localhost:%d/a/path", port)
+                    .to("mock:ce");
+            }
+        });
+
+        context.start();
+
+        MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
+        mock.expectedMessageCount(1);
+        mock.expectedHeaderReceived("CE-CloudEventsVersion", "0.1");
+        mock.expectedHeaderReceived("CE-EventType", "org.apache.camel.event");
+        mock.expectedHeaderReceived("CE-Source", "knative://endpoint/myEndpoint");
+        mock.expectedHeaderReceived(Exchange.CONTENT_TYPE, "text/plain");
+        mock.expectedMessagesMatches(e -> e.getIn().getHeaders().containsKey("CE-EventTime"));
+        mock.expectedMessagesMatches(e -> e.getIn().getHeaders().containsKey("CE-EventID"));
+        mock.expectedBodiesReceived("test");
+
+        context.createProducerTemplate().send(
+            "direct:source",
+            e -> {
+                e.getIn().setBody("test");
+            }
+        );
+
+        mock.assertIsSatisfied();
+    }
+
+    @Test
+    void testConsumeStructuredContent() throws Exception {
+        final int port = AvailablePortFinder.getNextAvailable();
+
+        KnativeEnvironment env = new KnativeEnvironment(Arrays.asList(
+            new KnativeEnvironment.KnativeServiceDefinition(
+                Knative.Type.endpoint,
+                Knative.Protocol.http,
+                "myEndpoint",
+                "localhost",
+                port,
+                mapOf(
+                    ServiceDefinition.SERVICE_META_PATH, "/a/path",
+                    Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+                    Knative.CONTENT_TYPE, "text/plain"
+                ))
+        ));
+
+        KnativeComponent component = context.getComponent("knative", KnativeComponent.class);
+        component.setEnvironment(env);
+
+        context.addRoutes(new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("knative:endpoint/myEndpoint")
+                    .to("mock:ce");
+
+                from("direct:source")
+                    .toF("undertow:<a href="http://localhost:%d/a/path">http://localhost:%d/a/path", port);
+            }
+        });
+
+        context.start();
+
+        MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
+        mock.expectedMessageCount(1);
+        mock.expectedHeaderReceived("CE-CloudEventsVersion", "0.1");
+        mock.expectedHeaderReceived("CE-EventType", "org.apache.camel.event");
+        mock.expectedHeaderReceived("CE-EventID", "myEventID");
+        mock.expectedHeaderReceived("CE-Source", "/somewhere");
+        mock.expectedHeaderReceived(Exchange.CONTENT_TYPE, "text/plain");
+        mock.expectedMessagesMatches(e -> e.getIn().getHeaders().containsKey("CE-EventTime"));
+        mock.expectedBodiesReceived("test");
+
+        context.createProducerTemplate().send(
+            "direct:source",
+            e -> {
+                e.getIn().setHeader(Exchange.CONTENT_TYPE, Knative.MIME_STRUCTURED_CONTENT_MODE);
+                e.getIn().setBody(new ObjectMapper().writeValueAsString(mapOf(
+                    "cloudEventsVersion", "0.1",
+                    "eventType", "org.apache.camel.event",
+                    "eventID", "myEventID",
+                    "eventTime", DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(ZonedDateTime.now()),
+                    "source", "/somewhere",
+                    "contentType", "text/plain",
+                    "data", "test"
+                )));
+            }
+        );
+
+        mock.assertIsSatisfied();
+    }
+
+    @Test
+    void testConsumeContent() throws Exception {
+        final int port = AvailablePortFinder.getNextAvailable();
+
+        KnativeEnvironment env = new KnativeEnvironment(Arrays.asList(
+            new KnativeEnvironment.KnativeServiceDefinition(
+                Knative.Type.endpoint,
+                Knative.Protocol.http,
+                "myEndpoint",
+                "localhost",
+                port,
+                mapOf(
+                    ServiceDefinition.SERVICE_META_PATH, "/a/path",
+                    Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+                    Knative.CONTENT_TYPE, "text/plain"
+                ))
+        ));
+
+        KnativeComponent component = context.getComponent("knative", KnativeComponent.class);
+        component.setEnvironment(env);
+
+        context.addRoutes(new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("knative:endpoint/myEndpoint")
+                    .to("mock:ce");
+
+                from("direct:source")
+                    .toF("undertow:<a href="http://localhost:%d/a/path">http://localhost:%d/a/path", port);
+            }
+        });
+
+        context.start();
+
+        MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
+        mock.expectedMessageCount(1);
+        mock.expectedHeaderReceived("CE-CloudEventsVersion", "0.1");
+        mock.expectedHeaderReceived("CE-EventType", "org.apache.camel.event");
+        mock.expectedHeaderReceived("CE-EventID", "myEventID");
+        mock.expectedHeaderReceived("CE-Source", "/somewhere");
+        mock.expectedHeaderReceived(Exchange.CONTENT_TYPE, "text/plain");
+        mock.expectedMessagesMatches(e -> e.getIn().getHeaders().containsKey("CE-EventTime"));
+        mock.expectedBodiesReceived("test");
+
+        context.createProducerTemplate().send(
+            "direct:source",
+            e -> {
+                e.getIn().setHeader(Exchange.CONTENT_TYPE, "text/plain");
+                e.getIn().setHeader("CE-CloudEventsVersion", "0.1");
+                e.getIn().setHeader("CE-EventType", "org.apache.camel.event");
+                e.getIn().setHeader("CE-EventID", "myEventID");
+                e.getIn().setHeader("CE-EventTime", DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(ZonedDateTime.now()));
+                e.getIn().setHeader("CE-Source", "/somewhere");
+                e.getIn().setBody("test");
+            }
+        );
+
+        mock.assertIsSatisfied();
+    }
+}
diff --git a/runtime/camel-knative/src/test/resources/environment.json b/runtime/camel-knative/src/test/resources/environment.json
new file mode 100644
index 0000000..4fbe864
--- /dev/null
+++ b/runtime/camel-knative/src/test/resources/environment.json
@@ -0,0 +1,26 @@
+{
+  "services": [
+    {
+      "type": "channel",
+      "protocol": "http",
+      "name": "c1",
+      "host": "localhost",
+      "port": "8001",
+      "metadata": {
+        "service.path": "",
+        "knative.event.type": ""
+      }
+    },
+    {
+      "type": "endpoint",
+      "protocol": "http",
+      "name": "e1",
+      "host": "localhost",
+      "port": "9001",
+      "metadata": {
+        "service.path": "",
+        "knative.event.type": ""
+      }
+    }
+  ]
+}
\ No newline at end of file
diff --git a/runtime/camel-knative/src/test/resources/log4j2.properties b/runtime/camel-knative/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..9d5f10e
--- /dev/null
+++ b/runtime/camel-knative/src/test/resources/log4j2.properties
@@ -0,0 +1,7 @@
+appender.console.type = Console
+appender.console.name = console
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = [%-5level] %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} - %msg%n
+
+rootLogger.level = INFO
+rootLogger.appenderRef.stdout.ref = console
\ No newline at end of file
diff --git a/runtime/pom.xml b/runtime/pom.xml
index 885a75e..4f05606 100644
--- a/runtime/pom.xml
+++ b/runtime/pom.xml
@@ -39,6 +39,7 @@
         <joor.version>0.9.9</joor.version>
         <commons-io.version>2.6</commons-io.version>
         <commons-lang.version>3.8.1</commons-lang.version>
+        <commons-collections4.version>4.2</commons-collections4.version>
         <assertj.version>3.11.1</assertj.version>
         <log4j2.version>2.11.1</log4j2.version>
         <slf4j.version>1.7.25</slf4j.version>
@@ -46,6 +47,7 @@
         <kotlin.version>1.3.0</kotlin.version>
         <snakeyaml.version>1.23</snakeyaml.version>
         <spock.version>1.0-groovy-2.4</spock.version>
+        <jackson.version>2.9.7</jackson.version>
 
         <gmavenplus-plugin.version>1.6.1</gmavenplus-plugin.version>
         <fabric8-maven-plugin.version>3.5.40</fabric8-maven-plugin.version>
@@ -78,6 +80,8 @@
                     <artifactId>maven-compiler-plugin</artifactId>
                     <version>3.8.0</version>
                     <configuration>
+                        <source>1.8</source>
+                        <target>1.8</target>
                         <compilerArgs>
                             <arg>-Xlint:deprecation</arg>
                         </compilerArgs>
@@ -93,6 +97,7 @@
         <module>kotlin</module>
         <module>catalog-builder</module>
         <module>dependency-lister</module>
+        <module>camel-knative</module>
     </modules>
 
 </project>