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 cb2b8934fbb7226e40c2ae965c84d9b768edc178
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Mon Sep 28 18:17:18 2020 +0200
Make kamelet:sink and kamelet:source work with workaround here (need to
find nicer solution in camel-core)
---
.../kamelet/KameletComponentConfigurer.java | 6 ++
.../kamelet/KameletEndpointConfigurer.java | 11 +++
.../apache/camel/component/kamelet/kamelet.json | 5 +
.../camel/component/kamelet/KameletComponent.java | 101 ++++++++++++++++++++-
.../camel/component/kamelet/KameletEndpoint.java | 3 +
5 files changed, 124 insertions(+), 2 deletions(-)
diff --git
a/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletComponentConfigurer.java
b/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletComponentConfigurer.java
index a0a384d..7c3717a 100644
---
a/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletComponentConfigurer.java
+++
b/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletComponentConfigurer.java
@@ -21,10 +21,12 @@ public class KameletComponentConfigurer extends
PropertyConfigurerSupport implem
switch (ignoreCase ? name.toLowerCase() : name) {
case "basicpropertybinding":
case "basicPropertyBinding":
target.setBasicPropertyBinding(property(camelContext, boolean.class, value));
return true;
+ case "block": target.setBlock(property(camelContext, boolean.class,
value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler":
target.setBridgeErrorHandler(property(camelContext, boolean.class, value));
return true;
case "lazystartproducer":
case "lazyStartProducer":
target.setLazyStartProducer(property(camelContext, boolean.class, value));
return true;
+ case "timeout": target.setTimeout(property(camelContext, long.class,
value)); return true;
default: return false;
}
}
@@ -33,8 +35,10 @@ public class KameletComponentConfigurer extends
PropertyConfigurerSupport implem
public Map<String, Object> getAllOptions(Object target) {
Map<String, Object> answer = new CaseInsensitiveMap();
answer.put("basicPropertyBinding", boolean.class);
+ answer.put("block", boolean.class);
answer.put("bridgeErrorHandler", boolean.class);
answer.put("lazyStartProducer", boolean.class);
+ answer.put("timeout", long.class);
return answer;
}
@@ -44,10 +48,12 @@ public class KameletComponentConfigurer extends
PropertyConfigurerSupport implem
switch (ignoreCase ? name.toLowerCase() : name) {
case "basicpropertybinding":
case "basicPropertyBinding": return target.isBasicPropertyBinding();
+ case "block": return target.isBlock();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "timeout": return target.getTimeout();
default: return null;
}
}
diff --git
a/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletEndpointConfigurer.java
b/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletEndpointConfigurer.java
index bb99f63..6869355 100644
---
a/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletEndpointConfigurer.java
+++
b/components/camel-kamelet/src/generated/java/org/apache/camel/component/kamelet/KameletEndpointConfigurer.java
@@ -21,15 +21,19 @@ public class KameletEndpointConfigurer extends
PropertyConfigurerSupport impleme
switch (ignoreCase ? name.toLowerCase() : name) {
case "basicpropertybinding":
case "basicPropertyBinding":
target.setBasicPropertyBinding(property(camelContext, boolean.class, value));
return true;
+ case "block": target.setBlock(property(camelContext, boolean.class,
value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler":
target.setBridgeErrorHandler(property(camelContext, boolean.class, value));
return true;
case "exceptionhandler":
case "exceptionHandler":
target.setExceptionHandler(property(camelContext,
org.apache.camel.spi.ExceptionHandler.class, value)); return true;
case "exchangepattern":
case "exchangePattern":
target.setExchangePattern(property(camelContext,
org.apache.camel.ExchangePattern.class, value)); return true;
+ case "kameletproperties":
+ case "kameletProperties":
target.setKameletProperties(property(camelContext, java.util.Map.class,
value)); return true;
case "lazystartproducer":
case "lazyStartProducer":
target.setLazyStartProducer(property(camelContext, boolean.class, value));
return true;
case "synchronous": target.setSynchronous(property(camelContext,
boolean.class, value)); return true;
+ case "timeout": target.setTimeout(property(camelContext, long.class,
value)); return true;
default: return false;
}
}
@@ -38,11 +42,14 @@ public class KameletEndpointConfigurer extends
PropertyConfigurerSupport impleme
public Map<String, Object> getAllOptions(Object target) {
Map<String, Object> answer = new CaseInsensitiveMap();
answer.put("basicPropertyBinding", boolean.class);
+ answer.put("block", boolean.class);
answer.put("bridgeErrorHandler", boolean.class);
answer.put("exceptionHandler",
org.apache.camel.spi.ExceptionHandler.class);
answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
+ answer.put("kameletProperties", java.util.Map.class);
answer.put("lazyStartProducer", boolean.class);
answer.put("synchronous", boolean.class);
+ answer.put("timeout", long.class);
return answer;
}
@@ -52,15 +59,19 @@ public class KameletEndpointConfigurer extends
PropertyConfigurerSupport impleme
switch (ignoreCase ? name.toLowerCase() : name) {
case "basicpropertybinding":
case "basicPropertyBinding": return target.isBasicPropertyBinding();
+ case "block": return target.isBlock();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "exceptionhandler":
case "exceptionHandler": return target.getExceptionHandler();
case "exchangepattern":
case "exchangePattern": return target.getExchangePattern();
+ case "kameletproperties":
+ case "kameletProperties": return target.getKameletProperties();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
case "synchronous": return target.isSynchronous();
+ case "timeout": return target.getTimeout();
default: return null;
}
}
diff --git
a/components/camel-kamelet/src/generated/resources/org/apache/camel/component/kamelet/kamelet.json
b/components/camel-kamelet/src/generated/resources/org/apache/camel/component/kamelet/kamelet.json
index 2da7754..a66f1ca 100644
---
a/components/camel-kamelet/src/generated/resources/org/apache/camel/component/kamelet/kamelet.json
+++
b/components/camel-kamelet/src/generated/resources/org/apache/camel/component/kamelet/kamelet.json
@@ -22,7 +22,9 @@
},
"componentProperties": {
"bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error
Handler", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions occurred while the
consumer is trying to pickup incoming messages, or the likes, will now be
processed as a message and handled by [...]
+ "block": { "kind": "property", "displayName": "Block", "group":
"producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "secret": false, "defaultValue":
"true", "description": "If sending a message to a kamelet endpoint which has no
active consumer, then we can tell the producer to block and wait for the
consumer to become active." },
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start
Producer", "group": "producer", "label": "producer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a producer may otherwise
fail during starting and cause the r [...]
+ "timeout": { "kind": "property", "displayName": "Timeout", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "secret": false, "defaultValue":
"30000", "description": "The timeout value to use if block is enabled." },
"basicPropertyBinding": { "kind": "property", "displayName": "Basic
Property Binding", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Whether the component should use basic
property binding (Camel 2.x) or the newer property binding with additional
capabilities" }
},
"properties": {
@@ -31,7 +33,10 @@
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error
Handler", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions occurred while the
consumer is trying to pickup incoming messages, or the likes, will now be
processed as a message and handled b [...]
"exceptionHandler": { "kind": "parameter", "displayName": "Exception
Handler", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "secret": false, "description": "To let the consumer use a
custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled
then this option is not in use. By default the consumer will deal with [...]
"exchangePattern": { "kind": "parameter", "displayName": "Exchange
Pattern", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut",
"InOptionalOut" ], "deprecated": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
+ "block": { "kind": "parameter", "displayName": "Block", "group":
"producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "secret": false, "defaultValue":
"true", "description": "If sending a message to a direct endpoint which has no
active consumer, then we can tell the producer to block and wait for the
consumer to become active." },
+ "kameletProperties": { "kind": "parameter", "displayName": "Kamelet
Properties", "group": "producer", "label": "producer", "required": false,
"type": "object", "javaType": "java.util.Map<java.lang.String,
java.lang.Object>", "deprecated": false, "secret": false, "defaultValue":
"true", "description": "Custom properties for kamelet" },
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start
Producer", "group": "producer", "label": "producer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a producer may otherwise
fail during starting and cause the [...]
+ "timeout": { "kind": "parameter", "displayName": "Timeout", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "secret": false, "defaultValue":
"30000", "description": "The timeout value to use if block is enabled." },
"basicPropertyBinding": { "kind": "parameter", "displayName": "Basic
Property Binding", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false,
"defaultValue": false, "description": "Whether the endpoint should use basic
property binding (Camel 2.x) or the newer property binding with additional
capabilities" },
"synchronous": { "kind": "parameter", "displayName": "Synchronous",
"group": "advanced", "label": "advanced", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "secret": false, "defaultValue":
"false", "description": "Sets whether synchronous processing should be strictly
used, or Camel is allowed to use asynchronous processing (if supported)." }
}
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
index 91a2514..1cc92b2 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
@@ -17,8 +17,12 @@
package org.apache.camel.component.kamelet;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
+import java.util.StringJoiner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -26,9 +30,16 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.VetoCamelContextStartException;
+import org.apache.camel.model.EndpointRequiredDefinition;
+import org.apache.camel.model.FromDefinition;
import org.apache.camel.model.ModelCamelContext;
+import org.apache.camel.model.ProcessorDefinitionHelper;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.RouteTemplateDefinition;
+import org.apache.camel.model.RouteTemplateParameterDefinition;
+import org.apache.camel.model.ToDefinition;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.PropertiesComponent;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
import org.apache.camel.support.LifecycleStrategySupport;
@@ -36,6 +47,9 @@ import org.apache.camel.support.service.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
+import static
org.apache.camel.model.ProcessorDefinitionHelper.filterTypeInOutputs;
+
/**
* The Kamelet Component provides support for materializing routes templates.
*/
@@ -57,6 +71,11 @@ public class KameletComponent extends DefaultComponent {
}
@Override
+ public Endpoint createEndpoint(String uri) throws Exception {
+ return super.createEndpoint(uri);
+ }
+
+ @Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
final String templateId = Kamelet.extractTemplateId(getCamelContext(),
remaining);
final String routeId = Kamelet.extractRouteId(getCamelContext(),
remaining);
@@ -209,14 +228,92 @@ public class KameletComponent extends DefaultComponent {
LOGGER.debug("Creating route from template={} and id={}",
endpoint.getTemplateId(), endpoint.getRouteId());
final ModelCamelContext context =
endpoint.getCamelContext().adapt(ModelCamelContext.class);
- final String id =
context.addRouteFromTemplate(endpoint.getRouteId(), endpoint.getTemplateId(),
endpoint.getKameletProperties());
+ final String id = addRouteFromTemplate(context,
endpoint.getRouteId(), endpoint.getTemplateId(),
endpoint.getKameletProperties());
final RouteDefinition def = context.getRouteDefinition(id);
if (!def.isPrepared()) {
- context.startRouteDefinitions(List.of(def));
+ // when starting the route that was created from the template
+ // then we must provide the route id as local properties to
the properties component
+ // as this route id is used internal by kamelets when they are
starting
+ PropertiesComponent pc = context.getPropertiesComponent();
+ try {
+ Properties prop = new Properties();
+ prop.put("routeId", id);
+ pc.setLocalProperties(prop);
+ context.startRouteDefinitions(List.of(def));
+ } finally {
+ pc.setLocalProperties(null);
+ }
}
LOGGER.debug("Route with id={} created from template={}", id,
endpoint.getTemplateId());
}
+
+ private static String addRouteFromTemplate(final ModelCamelContext
context, final String routeId, final String routeTemplateId, final Map<String,
Object> parameters)
+ throws Exception {
+ RouteTemplateDefinition target = null;
+ for (RouteTemplateDefinition def :
context.getRouteTemplateDefinitions()) {
+ if (routeTemplateId.equals(def.getId())) {
+ target = def;
+ break;
+ }
+ }
+ if (target == null) {
+ throw new IllegalArgumentException("Cannot find RouteTemplate
with id " + routeTemplateId);
+ }
+
+ StringJoiner templatesBuilder = new StringJoiner(", ");
+ final Map<String, Object> prop = new HashMap();
+ // include default values first from the template (and validate
that we have inputs for all required parameters)
+ if (target.getTemplateParameters() != null) {
+ for (RouteTemplateParameterDefinition temp :
target.getTemplateParameters()) {
+ if (temp.getDefaultValue() != null) {
+ prop.put(temp.getName(), temp.getDefaultValue());
+ } else {
+ // this is a required parameter do we have that as
input
+ if (!parameters.containsKey(temp.getName())) {
+ templatesBuilder.add(temp.getName());
+ }
+ }
+ }
+ }
+ if (templatesBuilder.length() > 0) {
+ throw new IllegalArgumentException(
+ "Route template " + routeTemplateId + " the following
mandatory parameters must be provided: "
+ + templatesBuilder.toString());
+ }
+ // then override with user parameters
+ if (parameters != null) {
+ prop.putAll(parameters);
+ }
+
+ RouteDefinition def = target.asRouteDefinition();
+ // must make deep copy of input
+ def.setInput(null);
+ def.setInput(new
FromDefinition(target.getRoute().getInput().getEndpointUri()));
+ if (routeId != null) {
+ def.setId(routeId);
+ }
+ // must make the source and simk endpoints are unique by appending
the route id before we create the route from the template
+ if (def.getInput().getEndpointUri().startsWith("kamelet:source")) {
+ def.getInput().setUri("kamelet:source?routeId=" + routeId);
+ }
+ Iterator<ToDefinition> it = filterTypeInOutputs(def.getOutputs(),
ToDefinition.class);
+ while (it.hasNext()) {
+ ToDefinition to = it.next();
+ if (to.getEndpointUri().startsWith("kamelet:sink")) {
+ // TODO: must make deep copy
+ to.setUri("kamelet:sink?routeId=" + routeId);
+ }
+ }
+
+
+ def.setTemplateParameters(prop);
+ context.removeRouteDefinition(def);
+ context.getRouteDefinitions().add(def);
+
+ return def.getId();
+ }
+
}
}
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
index 80d8e10..2d6e883 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
@@ -127,6 +127,9 @@ public class KameletEndpoint extends DefaultEndpoint {
return routeId;
}
+ /**
+ * Custom properties for kamelet
+ */
public void setKameletProperties(Map<String, Object> kameletProperties) {
if (kameletProperties != null) {
this.kameletProperties.clear();
