You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2020/06/22 09:53:00 UTC
[camel-k-runtime] branch master updated (c574262 -> c483481)
This is an automated email from the ASF dual-hosted git repository.
lburgazzoli pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k-runtime.git.
from c574262 Merge pull request #373 from apache/Surefire-3.0.0-M5
new 359c8e1 camel-knative: add support for lookup service definition from the registry #371
new ff5069c camel-knative: add support for URL in knative environment #369
new c483481 Add runtime support for Knative sinkbinding #365
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../knative/KnativeSourceLoaderInterceptor.java | 62 ++++-
.../knative/KnativeSourceRoutesLoaderTest.java | 95 +++++++-
.../src/test/resources/sources/routes.groovy | 2 +-
.../src/test/resources/sources/routes.java | 7 +-
.../src/test/resources/sources/routes.js | 2 +-
.../src/test/resources/sources/routes.kts | 2 +-
.../src/test/resources/sources/routes.xml | 2 +-
.../camel/component/knative/spi/Knative.java | 1 +
.../component/knative/spi/KnativeEnvironment.java | 265 +++++++++++++--------
.../knative/http/KnativeHttpProducer.java | 31 ++-
.../component/knative/http/KnativeHttpTest.java | 43 ++++
.../camel/component/knative/KnativeEndpoint.java | 69 +++---
.../component/knative/KnativeComponentTest.java | 10 +
13 files changed, 429 insertions(+), 162 deletions(-)
[camel-k-runtime] 03/03: Add runtime support for Knative
sinkbinding #365
Posted by lb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lburgazzoli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k-runtime.git
commit c48348130fafe949a6478e0b8511b6c84af90d10
Author: lburgazzoli <lb...@gmail.com>
AuthorDate: Sun Jun 21 13:51:36 2020 +0200
Add runtime support for Knative sinkbinding #365
---
.../knative/KnativeSourceLoaderInterceptor.java | 62 +++++++++++++-
.../knative/KnativeSourceRoutesLoaderTest.java | 95 ++++++++++++++++++++--
.../src/test/resources/sources/routes.groovy | 2 +-
.../src/test/resources/sources/routes.java | 7 +-
.../src/test/resources/sources/routes.js | 2 +-
.../src/test/resources/sources/routes.kts | 2 +-
.../src/test/resources/sources/routes.xml | 2 +-
.../component/knative/spi/KnativeEnvironment.java | 39 +++++++++
.../camel/component/knative/KnativeEndpoint.java | 2 +-
9 files changed, 189 insertions(+), 24 deletions(-)
diff --git a/camel-k-runtime-knative/src/main/java/org/apache/camel/k/loader/knative/KnativeSourceLoaderInterceptor.java b/camel-k-runtime-knative/src/main/java/org/apache/camel/k/loader/knative/KnativeSourceLoaderInterceptor.java
index bee157b..07f2113 100644
--- a/camel-k-runtime-knative/src/main/java/org/apache/camel/k/loader/knative/KnativeSourceLoaderInterceptor.java
+++ b/camel-k-runtime-knative/src/main/java/org/apache/camel/k/loader/knative/KnativeSourceLoaderInterceptor.java
@@ -16,17 +16,25 @@
*/
package org.apache.camel.k.loader.knative;
+import java.io.IOException;
+import java.io.Reader;
+import java.io.StringReader;
+import java.util.HashMap;
import java.util.List;
import java.util.Optional;
+import com.fasterxml.jackson.core.type.TypeReference;
import org.apache.camel.CamelContext;
import org.apache.camel.RoutesBuilder;
+import org.apache.camel.component.knative.spi.Knative;
+import org.apache.camel.component.knative.spi.KnativeEnvironment;
import org.apache.camel.k.Source;
import org.apache.camel.k.SourceLoader;
import org.apache.camel.k.annotation.LoaderInterceptor;
import org.apache.camel.k.support.RuntimeSupport;
import org.apache.camel.model.RouteDefinition;
import org.apache.camel.model.ToDefinition;
+import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,16 +54,22 @@ public class KnativeSourceLoaderInterceptor implements SourceLoader.Interceptor
return RuntimeSupport.afterConfigure(result.builder(), builder -> {
final CamelContext camelContext = builder.getContext();
final List<RouteDefinition> definitions = builder.getRouteCollection().getRoutes();
+
if (definitions.size() == 1) {
- final String sink = camelContext.resolvePropertyPlaceholders("{{env:KNATIVE_SINK:sink}}");
- final String uri = String.format("knative://endpoint/%s", sink);
+ final String sinkName = camelContext.resolvePropertyPlaceholders("{{knative.sink:sink}}");
+ final String sinkUri = String.format("knative://endpoint/%s", sinkName);
final RouteDefinition definition = definitions.get(0);
- LOGGER.info("Add sink:{} to route:{}", uri, definition.getId());
+ createSyntheticDefinition(camelContext, sinkName).ifPresent(serviceDefinition -> {
+ // publish the synthetic service definition
+ camelContext.getRegistry().bind(sinkName, serviceDefinition);
+ });
+
+ LOGGER.info("Add sink:{} to route:{}", sinkUri, definition.getId());
// assuming that route is linear like there's no content based routing
// or ant other EIP that would branch the flow
- definition.getOutputs().add(new ToDefinition(uri));
+ definition.getOutputs().add(new ToDefinition(sinkUri));
} else {
LOGGER.warn("Cannot determine route to enrich. the knative enpoint need to explicitly be defined");
}
@@ -68,4 +82,44 @@ public class KnativeSourceLoaderInterceptor implements SourceLoader.Interceptor
}
};
}
+
+ private static Optional<KnativeEnvironment.KnativeServiceDefinition> createSyntheticDefinition(
+ CamelContext camelContext,
+ String sinkName) {
+
+ final String kSinkUrl = camelContext.resolvePropertyPlaceholders("{{k.sink:}}");
+ final String kCeOverride = camelContext.resolvePropertyPlaceholders("{{k.ce.overrides:}}");
+
+ if (ObjectHelper.isNotEmpty(kSinkUrl)) {
+ // create a synthetic service definition to target the K_SINK url
+ var serviceBuilder = KnativeEnvironment.serviceBuilder(Knative.Type.endpoint, sinkName)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, Knative.EndpointKind.sink)
+ .withMeta(Knative.SERVICE_META_URL, kSinkUrl);
+
+ if (ObjectHelper.isNotEmpty(kCeOverride)) {
+ try (Reader reader = new StringReader(kCeOverride)) {
+ // assume K_CE_OVERRIDES is defined as simple key/val json
+ var overrides = Knative.MAPPER.readValue(
+ reader,
+ new TypeReference<HashMap<String, String>>() { }
+ );
+
+ for (var entry: overrides.entrySet()) {
+ // generate proper ce-override meta-data for the service
+ // definition
+ serviceBuilder.withMeta(
+ Knative.KNATIVE_CE_OVERRIDE_PREFIX + entry.getKey(),
+ entry.getValue()
+ );
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return Optional.of(serviceBuilder.build());
+ }
+
+ return Optional.empty();
+ }
}
diff --git a/camel-k-runtime-knative/src/test/java/org/apache/camel/k/loader/knative/KnativeSourceRoutesLoaderTest.java b/camel-k-runtime-knative/src/test/java/org/apache/camel/k/loader/knative/KnativeSourceRoutesLoaderTest.java
index e036299..7af4acd 100644
--- a/camel-k-runtime-knative/src/test/java/org/apache/camel/k/loader/knative/KnativeSourceRoutesLoaderTest.java
+++ b/camel-k-runtime-knative/src/test/java/org/apache/camel/k/loader/knative/KnativeSourceRoutesLoaderTest.java
@@ -18,7 +18,10 @@ package org.apache.camel.k.loader.knative;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import java.util.UUID;
import java.util.stream.Stream;
@@ -26,6 +29,9 @@ import org.apache.camel.CamelContext;
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.knative.KnativeComponent;
+import org.apache.camel.component.knative.KnativeConstants;
+import org.apache.camel.component.knative.spi.CloudEvent;
+import org.apache.camel.component.knative.spi.CloudEvents;
import org.apache.camel.component.knative.spi.Knative;
import org.apache.camel.component.knative.spi.KnativeEnvironment;
import org.apache.camel.component.mock.MockEndpoint;
@@ -74,11 +80,8 @@ public class KnativeSourceRoutesLoaderTest {
KnativeEnvironment.endpoint(Knative.EndpointKind.sink, "sink", "localhost", runtime.port)
));
-
CamelContext context = runtime.getCamelContext();
- context.disableJMX();
- context.setStreamCaching(true);
- context.addComponent("knative", component);
+ context.addComponent(KnativeConstants.SCHEME, component);
Source source = Sources.fromURI(uri);
SourceLoader loader = RoutesConfigurer.load(runtime, source);
@@ -112,11 +115,82 @@ public class KnativeSourceRoutesLoaderTest {
mock.expectedMessageCount(1);
mock.expectedBodiesReceived(data);
- context.createProducerTemplate().sendBodyAndHeader(
- "direct:start",
- "",
- "MyHeader",
- data);
+ context.createFluentProducerTemplate()
+ .to("direct:start")
+ .withHeader("MyHeader", data)
+ .send();
+
+ mock.assertIsSatisfied();
+ } finally {
+ context.stop();
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void testWrapLoaderWithSyntheticServiceDefinition(String uri) throws Exception {
+ LOGGER.info("uri: {}", uri);
+
+ final String data = UUID.randomUUID().toString();
+ final TestRuntime runtime = new TestRuntime();
+ final String typeHeaderKey = CloudEvents.v1_0.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http();
+ final String typeHeaderVal = UUID.randomUUID().toString();
+ final String url = String.format("http://localhost:%d", runtime.port);
+
+ KnativeComponent component = new KnativeComponent();
+ component.setEnvironment(new KnativeEnvironment(Collections.emptyList()));
+
+ Properties properties = new Properties();
+ properties.put("knative.sink", "mySynk");
+ properties.put("k.sink", String.format("http://localhost:%d", runtime.port));
+ properties.put("k.ce.overrides", Knative.MAPPER.writeValueAsString(Map.of(typeHeaderKey, typeHeaderVal)));
+
+ CamelContext context = runtime.getCamelContext();
+ context.getPropertiesComponent().setInitialProperties(properties);
+ context.addComponent(KnativeConstants.SCHEME, component);
+
+ Source source = Sources.fromURI(uri);
+ SourceLoader loader = RoutesConfigurer.load(runtime, source);
+
+ assertThat(loader.getSupportedLanguages()).contains(source.getLanguage());
+ assertThat(runtime.builders).hasSize(1);
+
+ try {
+ context.addRoutes(runtime.builders.get(0));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ fromF("platform-http:/")
+ .routeId("http")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ var definitions = context.adapt(ModelCamelContext.class).getRouteDefinitions();
+ var services = context.getRegistry().findByType(KnativeEnvironment.KnativeServiceDefinition.class);
+
+ assertThat(definitions).hasSize(2);
+ assertThat(definitions).first().satisfies(d -> {
+ assertThat(d.getOutputs()).last().hasFieldOrPropertyWithValue(
+ "endpointUri",
+ "knative://endpoint/mySynk"
+ );
+ });
+
+ assertThat(services).hasSize(1);
+ assertThat(services).first().hasFieldOrPropertyWithValue("name", "mySynk");
+ assertThat(services).first().hasFieldOrPropertyWithValue("url", url);
+
+ MockEndpoint mock = context.getEndpoint("mock:result", MockEndpoint.class);
+ mock.expectedMessageCount(1);
+ mock.expectedBodiesReceived(data);
+ mock.expectedHeaderReceived(typeHeaderKey, typeHeaderVal);
+
+ context.createFluentProducerTemplate()
+ .to("direct:start")
+ .withHeader("MyHeader", data)
+ .send();
mock.assertIsSatisfied();
} finally {
@@ -131,6 +205,9 @@ public class KnativeSourceRoutesLoaderTest {
public TestRuntime() {
this.camelContext = new DefaultCamelContext();
+ this.camelContext.disableJMX();
+ this.camelContext.setStreamCaching(true);
+
this.builders = new ArrayList<>();
this.port = AvailablePortFinder.getNextAvailable();
diff --git a/camel-k-runtime-knative/src/test/resources/sources/routes.groovy b/camel-k-runtime-knative/src/test/resources/sources/routes.groovy
index d4e7b78..6f1b3ce 100644
--- a/camel-k-runtime-knative/src/test/resources/sources/routes.groovy
+++ b/camel-k-runtime-knative/src/test/resources/sources/routes.groovy
@@ -15,5 +15,5 @@
* limitations under the License.
*/
from('direct:start')
- .setBody().simple('${header[MyHeader]}')
+ .setBody().header('MyHeader')
.to('log:knative')
\ No newline at end of file
diff --git a/camel-k-runtime-knative/src/test/resources/sources/routes.java b/camel-k-runtime-knative/src/test/resources/sources/routes.java
index ffa38de..14eb9e9 100644
--- a/camel-k-runtime-knative/src/test/resources/sources/routes.java
+++ b/camel-k-runtime-knative/src/test/resources/sources/routes.java
@@ -21,12 +21,7 @@ public class MyRoutes extends RouteBuilder {
@Override
public void configure() throws Exception {
from("direct:start")
- .setBody().simple("${header[MyHeader]}")
+ .setBody().header("MyHeader")
.to("log:knative");
}
-
- @BindToRegistry("my-bean")
- public static String myBean() {
- return "my-bean-string";
- }
}
\ No newline at end of file
diff --git a/camel-k-runtime-knative/src/test/resources/sources/routes.js b/camel-k-runtime-knative/src/test/resources/sources/routes.js
index 75de4de..041e045 100644
--- a/camel-k-runtime-knative/src/test/resources/sources/routes.js
+++ b/camel-k-runtime-knative/src/test/resources/sources/routes.js
@@ -15,5 +15,5 @@
* limitations under the License.
*/
from('direct:start')
- .setBody().simple('${header[MyHeader]}')
+ .setBody().header('MyHeader')
.to('log:knative');
\ No newline at end of file
diff --git a/camel-k-runtime-knative/src/test/resources/sources/routes.kts b/camel-k-runtime-knative/src/test/resources/sources/routes.kts
index 48b8b1c..54cc88c 100644
--- a/camel-k-runtime-knative/src/test/resources/sources/routes.kts
+++ b/camel-k-runtime-knative/src/test/resources/sources/routes.kts
@@ -15,6 +15,6 @@
* limitations under the License.
*/
from("direct:start")
- .setBody().simple("\${header[MyHeader]}")
+ .setBody().header("MyHeader")
.to("log:knative")
diff --git a/camel-k-runtime-knative/src/test/resources/sources/routes.xml b/camel-k-runtime-knative/src/test/resources/sources/routes.xml
index f0d953a..a7daa07 100644
--- a/camel-k-runtime-knative/src/test/resources/sources/routes.xml
+++ b/camel-k-runtime-knative/src/test/resources/sources/routes.xml
@@ -21,7 +21,7 @@
<route>
<from uri="direct:start"/>
<setBody>
- <simple>${header[MyHeader]}</simple>
+ <header>MyHeader</header>
</setBody>
<to uri="log:knative"/>
</route>
diff --git a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
index 2884dc9..5c9b7a8 100644
--- a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
+++ b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
@@ -19,6 +19,8 @@ package org.apache.camel.component.knative.spi;
import java.io.InputStream;
import java.io.Reader;
import java.io.StringReader;
+import java.net.MalformedURLException;
+import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -263,11 +265,48 @@ public class KnativeEnvironment {
);
}
+ @Override
+ public String getHost() {
+ String urlAsString = getUrl();
+ if (urlAsString != null) {
+ try {
+ return new URL(urlAsString).getHost();
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return super.getHost();
+ }
+
+ @Override
+ public int getPort() {
+ String urlAsString = getUrl();
+ if (urlAsString != null) {
+ try {
+ return new URL(urlAsString).getPort();
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return super.getPort();
+ }
+
public Knative.Type getType() {
return Knative.Type.valueOf(getMetadata().get(Knative.KNATIVE_TYPE));
}
public String getPath() {
+ String urlAsString = getUrl();
+ if (urlAsString != null) {
+ try {
+ return new URL(urlAsString).getPath();
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
return getMetadata(Knative.SERVICE_META_PATH);
}
diff --git a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
index 4b989bf..f02f1f5 100644
--- a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
+++ b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
@@ -139,7 +139,7 @@ public class KnativeEndpoint extends DefaultEndpoint {
//
KnativeEnvironment.KnativeServiceDefinition service = lookupServiceDefinition(serviceName, endpointKind)
.or(() -> lookupServiceDefinition("default", endpointKind))
- .orElseThrow(() -> new IllegalArgumentException(String.format("Unable to find a service definition for %s/%s/%s", type, serviceName, endpointKind)));
+ .orElseThrow(() -> new IllegalArgumentException(String.format("Unable to find a service definition for %s/%s/%s", type, endpointKind, serviceName)));
final Map<String, String> metadata = new HashMap<>(service.getMetadata());
[camel-k-runtime] 01/03: camel-knative: add support for lookup
service definition from the registry #371
Posted by lb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lburgazzoli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k-runtime.git
commit 359c8e1ccd4b93e66df092b02e67bd52a3cdd9ab
Author: lburgazzoli <lb...@gmail.com>
AuthorDate: Sun Jun 21 11:09:47 2020 +0200
camel-knative: add support for lookup service definition from the registry #371
---
.../component/knative/spi/KnativeEnvironment.java | 15 +++--
.../camel/component/knative/KnativeEndpoint.java | 67 +++++++++++++---------
.../component/knative/KnativeComponentTest.java | 10 ++++
3 files changed, 59 insertions(+), 33 deletions(-)
diff --git a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
index ee43bbe..76af6de 100644
--- a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
+++ b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
@@ -53,11 +53,7 @@ public class KnativeEnvironment {
}
public Stream<KnativeServiceDefinition> lookup(Knative.Type type, String name) {
- return services.stream()
- .filter(definition -> {
- return Objects.equals(type.name(), definition.getMetadata().get(Knative.KNATIVE_TYPE))
- && Objects.equals(name, definition.getName());
- });
+ return stream().filter(definition -> definition.matches(type, name));
}
// ************************
@@ -267,5 +263,14 @@ public class KnativeEnvironment {
public int getPortOrDefault(int port) {
return getPort() != -1 ? getPort() : port;
}
+
+ public String getMetadata(String key) {
+ return getMetadata().get(key);
+ }
+
+ public boolean matches(Knative.Type type, String name) {
+ return Objects.equals(type.name(), getMetadata(Knative.KNATIVE_TYPE))
+ && Objects.equals(name, getName());
+ }
}
}
diff --git a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
index ed6674b..4b989bf 100644
--- a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
+++ b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/KnativeEndpoint.java
@@ -20,6 +20,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
@@ -51,12 +53,10 @@ public class KnativeEndpoint extends DefaultEndpoint {
private final Knative.Type type;
@UriPath(description = "The Knative name")
private final String name;
-
+ private final CloudEventProcessor cloudEvent;
@UriParam
private KnativeConfiguration configuration;
- private final CloudEventProcessor cloudEvent;
-
public KnativeEndpoint(String uri, KnativeComponent component, Knative.Type type, String name, KnativeConfiguration configuration) {
super(uri, component);
@@ -120,14 +120,14 @@ public class KnativeEndpoint extends DefaultEndpoint {
return name;
}
- public void setConfiguration(KnativeConfiguration configuration) {
- this.configuration = configuration;
- }
-
public KnativeConfiguration getConfiguration() {
return configuration;
}
+ public void setConfiguration(KnativeConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
KnativeEnvironment.KnativeServiceDefinition lookupServiceDefinition(Knative.EndpointKind endpointKind) {
String serviceName = configuration.getServiceName();
@@ -152,7 +152,7 @@ public class KnativeEndpoint extends DefaultEndpoint {
key = Knative.KNATIVE_FILTER_PREFIX + key;
}
- metadata.put(key, (String)val);
+ metadata.put(key, (String) val);
}
}
@@ -165,7 +165,7 @@ public class KnativeEndpoint extends DefaultEndpoint {
key = Knative.KNATIVE_CE_OVERRIDE_PREFIX + key;
}
- metadata.put(key, (String)val);
+ metadata.put(key, (String) val);
}
}
@@ -184,25 +184,9 @@ public class KnativeEndpoint extends DefaultEndpoint {
}
Optional<KnativeEnvironment.KnativeServiceDefinition> lookupServiceDefinition(String name, Knative.EndpointKind endpointKind) {
- return this.configuration.getEnvironment()
- .lookup(this.type, name)
- .filter(s -> {
- final String type = s.getMetadata().get(Knative.CAMEL_ENDPOINT_KIND);
- final String apiv = s.getMetadata().get(Knative.KNATIVE_API_VERSION);
- final String kind = s.getMetadata().get(Knative.KNATIVE_KIND);
-
- if (!Objects.equals(endpointKind.name(), type)) {
- return false;
- }
- if (configuration.getApiVersion() != null && !Objects.equals(apiv, configuration.getApiVersion())) {
- return false;
- }
- if (configuration.getKind() != null && !Objects.equals(kind, configuration.getKind())) {
- return false;
- }
-
- return true;
- })
+ return servicesDefinitions()
+ .filter(definition -> definition.matches(this.type, name))
+ .filter(serviceFilter(endpointKind))
.findFirst();
}
@@ -212,4 +196,31 @@ public class KnativeEndpoint extends DefaultEndpoint {
!this.configuration.isReplyWithCloudEvent()
);
}
+
+ private Stream<KnativeEnvironment.KnativeServiceDefinition> servicesDefinitions() {
+ return Stream.concat(
+ getCamelContext().getRegistry().findByType(KnativeEnvironment.KnativeServiceDefinition.class).stream(),
+ this.configuration.getEnvironment().stream()
+ );
+ }
+
+ private Predicate<KnativeEnvironment.KnativeServiceDefinition> serviceFilter(Knative.EndpointKind endpointKind) {
+ return s -> {
+ final String type = s.getMetadata(Knative.CAMEL_ENDPOINT_KIND);
+ final String apiv = s.getMetadata(Knative.KNATIVE_API_VERSION);
+ final String kind = s.getMetadata(Knative.KNATIVE_KIND);
+
+ if (!Objects.equals(endpointKind.name(), type)) {
+ return false;
+ }
+ if (configuration.getApiVersion() != null && !Objects.equals(apiv, configuration.getApiVersion())) {
+ return false;
+ }
+ if (configuration.getKind() != null && !Objects.equals(kind, configuration.getKind())) {
+ return false;
+ }
+
+ return true;
+ };
+ }
}
diff --git a/camel-knative/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java b/camel-knative/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java
index 7a1870c..ac6ed7a 100644
--- a/camel-knative/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java
+++ b/camel-knative/camel-knative/src/test/java/org/apache/camel/component/knative/KnativeComponentTest.java
@@ -67,6 +67,8 @@ public class KnativeComponentTest {
component.setEnvironment(env);
component.setTransport(new KnativeTransportNoop());
+ context.getRegistry().bind("ereg", KnativeEnvironment.endpoint(Knative.EndpointKind.source, "ereg", null, -1));
+ context.getRegistry().bind("creg", KnativeEnvironment.channel(Knative.EndpointKind.source, "creg", null, -1));
context.addComponent("knative", component);
//
@@ -77,6 +79,10 @@ public class KnativeComponentTest {
assertThat(endpoint.lookupServiceDefinition("c1", Knative.EndpointKind.source)).isPresent();
assertThat(endpoint.lookupServiceDefinition("e1", Knative.EndpointKind.source)).isNotPresent();
}
+ {
+ KnativeEndpoint endpoint = context.getEndpoint("knative:channel/creg", KnativeEndpoint.class);
+ assertThat(endpoint.lookupServiceDefinition("creg", Knative.EndpointKind.source)).isPresent();
+ }
//
// Endpoints
@@ -86,5 +92,9 @@ public class KnativeComponentTest {
assertThat(endpoint.lookupServiceDefinition("e1", Knative.EndpointKind.source)).isPresent();
assertThat(endpoint.lookupServiceDefinition("c1", Knative.EndpointKind.source)).isNotPresent();
}
+ {
+ KnativeEndpoint endpoint = context.getEndpoint("knative:endpoint/ereg", KnativeEndpoint.class);
+ assertThat(endpoint.lookupServiceDefinition("ereg", Knative.EndpointKind.source)).isPresent();
+ }
}
}
[camel-k-runtime] 02/03: camel-knative: add support for URL in
knative environment #369
Posted by lb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lburgazzoli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k-runtime.git
commit ff5069c0de9adaa6d123c1a81666ffcefe59d4dc
Author: lburgazzoli <lb...@gmail.com>
AuthorDate: Sun Jun 21 11:55:05 2020 +0200
camel-knative: add support for URL in knative environment #369
---
.../camel/component/knative/spi/Knative.java | 1 +
.../component/knative/spi/KnativeEnvironment.java | 211 ++++++++++++---------
.../knative/http/KnativeHttpProducer.java | 31 +--
.../component/knative/http/KnativeHttpTest.java | 43 +++++
4 files changed, 181 insertions(+), 105 deletions(-)
diff --git a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/Knative.java b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/Knative.java
index 4d52932..a9dbbc0 100644
--- a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/Knative.java
+++ b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/Knative.java
@@ -37,6 +37,7 @@ public final class Knative {
public static final String SERVICE_META_HOST = "service.host";
public static final String SERVICE_META_ZONE = "service.zone";
public static final String SERVICE_META_PATH = "service.path";
+ public static final String SERVICE_META_URL = "service.url";
private Knative() {
}
diff --git a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
index 76af6de..2884dc9 100644
--- a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
+++ b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/KnativeEnvironment.java
@@ -21,7 +21,7 @@ import java.io.Reader;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -33,8 +33,6 @@ import org.apache.camel.CamelContext;
import org.apache.camel.impl.cloud.DefaultServiceDefinition;
import org.apache.camel.support.ResourceHelper;
-import static org.apache.camel.util.CollectionHelper.mapOf;
-
/*
* Assuming it is loaded from a json for now
*/
@@ -70,7 +68,6 @@ public class KnativeEnvironment {
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. :
//
@@ -100,129 +97,151 @@ public class KnativeEnvironment {
}
public static KnativeServiceDefinition endpoint(Knative.EndpointKind endpointKind, String name, String host, int port) {
- return entry(
- endpointKind,
- Knative.Type.endpoint,
- name,
- host,
- port,
- Collections.emptyMap()
- );
+ return serviceBuilder(Knative.Type.endpoint, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeServiceDefinition endpoint(Knative.EndpointKind endpointKind, String name, String host, int port, Map<String, String> metadata) {
- return entry(
- endpointKind,
- Knative.Type.endpoint,
- name,
- host,
- port,
- metadata
- );
+ return serviceBuilder(Knative.Type.endpoint, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(metadata)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeServiceDefinition sourceEndpoint(String name, Map<String, String> metadata) {
- return entry(
- Knative.EndpointKind.source,
- Knative.Type.endpoint,
- name,
- null,
- -1,
- metadata
- );
+ return serviceBuilder(Knative.Type.endpoint, name)
+ .withMeta(metadata)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, Knative.EndpointKind.source.name())
+ .build();
}
public static KnativeServiceDefinition channel(Knative.EndpointKind endpointKind, String name, String host, int port) {
- return entry(
- endpointKind,
- Knative.Type.channel,
- name,
- host,
- port,
- Collections.emptyMap()
- );
+ return serviceBuilder(Knative.Type.channel, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeServiceDefinition channel(Knative.EndpointKind endpointKind, String name, String host, int port, Map<String, String> metadata) {
- return entry(
- endpointKind,
- Knative.Type.channel,
- name,
- host,
- port,
- metadata
- );
+ return serviceBuilder(Knative.Type.channel, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(metadata)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeServiceDefinition event(Knative.EndpointKind endpointKind, String name, String host, int port) {
- return entry(
- endpointKind,
- Knative.Type.event,
- name,
- host,
- port,
- Collections.emptyMap()
- );
+ return serviceBuilder(Knative.Type.event, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeServiceDefinition sourceEvent(String name) {
- return entry(
- Knative.EndpointKind.source,
- Knative.Type.event,
- name,
- null,
- -1,
- Collections.emptyMap()
- );
+ return serviceBuilder(Knative.Type.event, name)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, Knative.EndpointKind.source)
+ .build();
}
public static KnativeServiceDefinition sourceEvent(String name, Map<String, String> metadata) {
- return entry(
- Knative.EndpointKind.source,
- Knative.Type.event,
- name,
- null,
- -1,
- metadata
- );
+ return serviceBuilder(Knative.Type.event, name)
+ .withMeta(metadata)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, Knative.EndpointKind.source)
+ .build();
}
public static KnativeServiceDefinition event(Knative.EndpointKind endpointKind, String name, String host, int port, Map<String, String> metadata) {
- return entry(
- endpointKind,
- Knative.Type.event,
- name,
- host,
- port,
- metadata
- );
- }
-
- public static KnativeServiceDefinition entry(Knative.EndpointKind endpointKind, Knative.Type type, String name, String host, int port, Map<String, String> metadata) {
- return new KnativeEnvironment.KnativeServiceDefinition(
- type,
- name,
- host,
- port,
- KnativeSupport.mergeMaps(
- metadata,
- mapOf(
- Knative.CAMEL_ENDPOINT_KIND, endpointKind.name()
- )
- )
- );
+ return serviceBuilder(Knative.Type.event, name)
+ .withHost(host)
+ .withPort(port)
+ .withMeta(metadata)
+ .withMeta(Knative.CAMEL_ENDPOINT_KIND, endpointKind)
+ .build();
}
public static KnativeEnvironment on(KnativeServiceDefinition... definitions) {
return new KnativeEnvironment(Arrays.asList(definitions));
}
+ public static KnativeServiceBuilder serviceBuilder(Knative.Type type, String name) {
+ return new KnativeServiceBuilder(type, name);
+ }
+
// ************************
//
// Types
//
// ************************
+
+ public static final class KnativeServiceBuilder {
+ private final Knative.Type type;
+ private final String name;
+ private String host;
+ private Integer port;
+ private Map<String, String> metadata;
+
+ public KnativeServiceBuilder(Knative.Type type, String name) {
+ this.type = type;
+ this.name = name;
+ }
+
+ public KnativeServiceBuilder withHost(String host) {
+ this.host = host;
+ return this;
+ }
+
+ public KnativeServiceBuilder withPort(Integer port) {
+ this.port = port;
+ return this;
+ }
+
+ public KnativeServiceBuilder withMeta(Map<String, String> metadata) {
+ if (metadata == null) {
+ return this;
+ }
+ if (this.metadata == null) {
+ this.metadata = new HashMap<>();
+ }
+ this.metadata.putAll(metadata);
+ return this;
+ }
+
+ public KnativeServiceBuilder withMeta(String key, String value) {
+ if (key == null || value == null) {
+ return this;
+ }
+ if (this.metadata == null) {
+ this.metadata = new HashMap<>();
+ }
+ this.metadata.put(key, value);
+ return this;
+ }
+
+ public KnativeServiceBuilder withMeta(String key, Enum<?> e) {
+ if (key == null || e == null) {
+ return this;
+ }
+ if (this.metadata == null) {
+ this.metadata = new HashMap<>();
+ }
+ this.metadata.put(key, e.name());
+ return this;
+ }
+
+ public KnativeServiceDefinition build() {
+ return new KnativeServiceDefinition(type, name, host, port, metadata);
+ }
+ }
+
public static final class KnativeServiceDefinition extends DefaultServiceDefinition {
@JsonCreator
public KnativeServiceDefinition(
@@ -238,7 +257,7 @@ public class KnativeEnvironment {
port == null ? -1 : port,
KnativeSupport.mergeMaps(
metadata,
- mapOf(
+ Map.of(
Knative.KNATIVE_TYPE, type.name())
)
);
@@ -249,7 +268,7 @@ public class KnativeEnvironment {
}
public String getPath() {
- return getMetadata().get(Knative.SERVICE_META_PATH);
+ return getMetadata(Knative.SERVICE_META_PATH);
}
public String getPathOrDefault(String path) {
@@ -257,13 +276,17 @@ public class KnativeEnvironment {
}
public String getEventType() {
- return getMetadata().get(Knative.KNATIVE_EVENT_TYPE);
+ return getMetadata(Knative.KNATIVE_EVENT_TYPE);
}
public int getPortOrDefault(int port) {
return getPort() != -1 ? getPort() : port;
}
+ public String getUrl() {
+ return getMetadata(Knative.SERVICE_META_URL);
+ }
+
public String getMetadata(String key) {
return getMetadata().get(key);
}
diff --git a/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpProducer.java b/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpProducer.java
index d707a51..010bfb3 100644
--- a/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpProducer.java
+++ b/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpProducer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.knative.http;
import java.util.Map;
+import java.util.function.Supplier;
import io.vertx.core.MultiMap;
import io.vertx.core.Vertx;
@@ -38,6 +39,7 @@ import org.apache.camel.support.DefaultMessage;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.URISupport;
+import org.apache.camel.util.function.Suppliers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +51,7 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
private final Vertx vertx;
private final WebClientOptions clientOptions;
private final HeaderFilterStrategy headerFilterStrategy;
+ private final Supplier<String> uri;
private WebClient client;
@@ -65,6 +68,7 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
this.vertx = ObjectHelper.notNull(vertx, "vertx");
this.clientOptions = ObjectHelper.supplyIfEmpty(clientOptions, WebClientOptions::new);
this.headerFilterStrategy = new KnativeHttpHeaderFilterStrategy();
+ this.uri = Suppliers.memorize(() -> computeUrl(serviceDefinition));
}
@Override
@@ -111,10 +115,7 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
return true;
}
- final int port = serviceDefinition.getPortOrDefault(KnativeHttpTransport.DEFAULT_PORT);
- final String path = serviceDefinition.getPathOrDefault(KnativeHttpTransport.DEFAULT_PATH);
-
- client.post(port, serviceDefinition.getHost(), path)
+ client.postAbs(this.uri.get())
.putHeaders(headers)
.sendBuffer(Buffer.buffer(payload), response -> {
if (response.succeeded()) {
@@ -136,7 +137,7 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
if (result.statusCode() < 200 || result.statusCode() >= 300) {
String exceptionMessage = String.format(
"HTTP operation failed invoking %s with statusCode: %d, statusMessage: %s",
- URISupport.sanitizeUri(getURI()),
+ URISupport.sanitizeUri(this.uri.get()),
result.statusCode(),
result.statusMessage()
);
@@ -148,7 +149,7 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
exchange.setMessage(answer);
} else if (response.failed()) {
- String exceptionMessage = "HTTP operation failed invoking " + URISupport.sanitizeUri(getURI());
+ String exceptionMessage = "HTTP operation failed invoking " + URISupport.sanitizeUri(this.uri.get());
if (response.result() != null) {
exceptionMessage += " with statusCode: " + response.result().statusCode();
}
@@ -180,12 +181,20 @@ public class KnativeHttpProducer extends DefaultAsyncProducer {
}
}
- private String getURI() {
- String p = ObjectHelper.supplyIfEmpty(serviceDefinition.getPath(), () -> KnativeHttpTransport.DEFAULT_PATH);
- if (!p.startsWith("/")) {
- p = "/" + p;
+ private static String computeUrl(KnativeEnvironment.KnativeServiceDefinition definition) {
+ String url = definition.getUrl();
+ if (url == null) {
+ int port = definition.getPortOrDefault(KnativeHttpTransport.DEFAULT_PORT);
+ String path = definition.getPathOrDefault(KnativeHttpTransport.DEFAULT_PATH);
+
+ if (!path.startsWith("/")) {
+ path = "/" + path;
+ }
+
+ url = String.format("http://%s:%d%s", definition.getHost(), port, path);
}
- return String.format("http://%s:%d%s", serviceDefinition.getHost(), serviceDefinition.getPort(), p);
+ return url;
}
+
}
diff --git a/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java b/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
index edb81cc..a896c1c 100644
--- a/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
+++ b/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
@@ -234,6 +234,49 @@ public class KnativeHttpTest {
@ParameterizedTest
@EnumSource(CloudEvents.class)
+ void testInvokeEndpointByUrl(CloudEvent ce) throws Exception {
+ configureKnativeComponent(
+ context,
+ ce,
+ endpoint(
+ Knative.EndpointKind.sink,
+ "myEndpoint",
+ "none",
+ -1,
+ mapOf(
+ Knative.SERVICE_META_PATH, "/does/not/exist",
+ Knative.SERVICE_META_URL, String.format("http://localhost:%d/a/path", platformHttpPort),
+ Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+ Knative.CONTENT_TYPE, "text/plain"
+ ))
+ );
+
+ RouteBuilder.addRoutes(context, b -> {
+ b.from("direct:source")
+ .to("knative:endpoint/myEndpoint");
+ b.from("platform-http:/a/path")
+ .to("mock:ce");
+ });
+
+ context.start();
+
+ MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).http(), ce.version());
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http(), "org.apache.camel.event");
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).http(), "knative://endpoint/myEndpoint");
+ mock.expectedHeaderReceived(Exchange.CONTENT_TYPE, "text/plain");
+ mock.expectedMessagesMatches(e -> e.getMessage().getHeaders().containsKey(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TIME).http()));
+ mock.expectedMessagesMatches(e -> e.getMessage().getHeaders().containsKey(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).http()));
+ mock.expectedBodiesReceived("test");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:source", "test");
+
+ mock.assertIsSatisfied();
+ }
+
+ @ParameterizedTest
+ @EnumSource(CloudEvents.class)
void testConsumeStructuredContent(CloudEvent ce) throws Exception {
configureKnativeComponent(