This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 1f491e25ff4 CAMEL-21660: camel-kamelet - Allow to configure consumer
bridgeErrorHandler option
1f491e25ff4 is described below
commit 1f491e25ff44bd78f1b2a85ed8f1dfc9bdf69ce7
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jan 27 14:38:19 2025 +0100
CAMEL-21660: camel-kamelet - Allow to configure consumer bridgeErrorHandler
option
---
.../apache/camel/component/kamelet/Kamelet.java | 14 ++
.../camel/component/kamelet/KameletComponent.java | 2 +
.../camel/component/kamelet/KameletEndpoint.java | 9 +-
.../KameletSourceBridgeErrorHandlerTest.java | 156 +++++++++++++++++++++
4 files changed, 180 insertions(+), 1 deletion(-)
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java
index 5e7c905ad4c..f9a642074df 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/Kamelet.java
@@ -54,6 +54,7 @@ public final class Kamelet {
public static final String PARENT_ROUTE_ID = "parentRouteId";
public static final String PARENT_PROCESSOR_ID = "parentProcessorId";
public static final String NO_ERROR_HANDLER = "noErrorHandler";
+ public static final String BRIDGE_ERROR_HANDLER = "bridgeErrorHandler";
// use a running counter as uuid
private static final UuidGenerator UUID = new SimpleUuidGenerator();
@@ -186,6 +187,7 @@ public final class Kamelet {
public static RouteDefinition templateToRoute(RouteTemplateDefinition in,
Map<String, Object> parameters) {
final String rid = (String) parameters.get(PARAM_ROUTE_ID);
final boolean noErrorHandler = (boolean)
parameters.get(NO_ERROR_HANDLER);
+ final boolean bridgeErrorHandler = (boolean)
parameters.get(BRIDGE_ERROR_HANDLER);
final String uuid = (String) parameters.get(PARAM_UUID);
final String prid = (String) parameters.get(PARENT_ROUTE_ID);
final String ppid = (String) parameters.get(PARENT_PROCESSOR_ID);
@@ -232,6 +234,18 @@ public final class Kamelet {
def.getInput().setUri("kamelet://source?" + PARAM_ROUTE_ID + "=" +
rid);
}
+ if (bridgeErrorHandler) {
+ String uri = def.getInput().getUri();
+ if (!uri.contains("bridgeErrorHandler=")) {
+ if (uri.contains("?")) {
+ uri = uri + "&bridgeErrorHandler=true";
+ } else {
+ uri = uri + "?bridgeErrorHandler=true";
+ }
+ def.getInput().setUri(uri);
+ }
+ }
+
// there must be at least one sink
int line = -1;
boolean sink = false;
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 97711779607..af5265df24b 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
@@ -49,6 +49,7 @@ import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.camel.component.kamelet.Kamelet.BRIDGE_ERROR_HANDLER;
import static org.apache.camel.component.kamelet.Kamelet.NO_ERROR_HANDLER;
import static org.apache.camel.component.kamelet.Kamelet.PARAM_LOCATION;
import static org.apache.camel.component.kamelet.Kamelet.PARAM_ROUTE_ID;
@@ -247,6 +248,7 @@ public class KameletComponent extends DefaultComponent {
kameletProperties.put(PARAM_ROUTE_ID, routeId);
kameletProperties.put(PARAM_UUID, uuid);
kameletProperties.put(NO_ERROR_HANDLER,
endpoint.isNoErrorHandler());
+ kameletProperties.put(BRIDGE_ERROR_HANDLER,
endpoint.isBridgeErrorHandler());
// set kamelet specific properties
endpoint.setKameletProperties(kameletProperties);
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 8481f3eeaa1..f92dedfea86 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
@@ -20,6 +20,8 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.apache.camel.AfterPropertiesConfigured;
+import org.apache.camel.CamelContext;
import org.apache.camel.Category;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
@@ -35,7 +37,7 @@ import org.apache.camel.util.ObjectHelper;
@UriEndpoint(firstVersion = "3.8.0", scheme = "kamelet", syntax =
"kamelet:templateId/routeId", title = "Kamelet",
lenientProperties = true, category = Category.CORE)
-public class KameletEndpoint extends DefaultEndpoint {
+public class KameletEndpoint extends DefaultEndpoint implements
AfterPropertiesConfigured {
private final String key;
private final Map<String, Object> kameletProperties;
@@ -186,6 +188,11 @@ public class KameletEndpoint extends DefaultEndpoint {
return answer;
}
+ @Override
+ public void afterPropertiesConfigured(CamelContext camelContext) {
+ kameletProperties.put(Kamelet.BRIDGE_ERROR_HANDLER,
isBridgeErrorHandler());
+ }
+
@Override
public void setProperties(Object bean, Map<String, Object> parameters) {
if (parameters == null || parameters.isEmpty()) {
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletSourceBridgeErrorHandlerTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletSourceBridgeErrorHandlerTest.java
new file mode 100644
index 00000000000..0e1e251f1ec
--- /dev/null
+++
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletSourceBridgeErrorHandlerTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.kamelet;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class KameletSourceBridgeErrorHandlerTest extends CamelTestSupport {
+
+ protected final CountDownLatch latch = new CountDownLatch(1);
+
+ @Test
+ public void testBridgeErrorHandler() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World",
"Hello World");
+ getMockEndpoint("mock:dead").expectedBodiesReceived("Cannot process");
+
+ latch.countDown();
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ Exception cause =
getMockEndpoint("mock:dead").getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT,
+ Exception.class);
+ assertNotNull(cause);
+ assertEquals("Simulated", cause.getMessage());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ getContext().addComponent("my", new MyComponent());
+
+ routeTemplate("foo")
+ .from("my:foo")
+ .to("mock:foo");
+
+ errorHandler(deadLetterChannel("mock:dead"));
+
+ from("kamelet:foo?bridgeErrorHandler=true").routeId("start")
+ .to("mock:result");
+ }
+ };
+ }
+
+ public class MyComponent extends DefaultComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new MyEndpoint(uri, this);
+ }
+ }
+
+ public class MyEndpoint extends DefaultEndpoint {
+
+ public MyEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ return null;
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ Consumer answer = new MyConsumer(this, processor);
+ configureConsumer(answer);
+ return answer;
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+ }
+
+ public class MyConsumer extends DefaultConsumer {
+
+ private int invoked;
+
+ public MyConsumer(Endpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+
+ public void doSomething() {
+ try {
+ if (invoked++ == 0) {
+ throw new IllegalArgumentException("Simulated");
+ }
+
+ Exchange exchange = getEndpoint().createExchange();
+ exchange.getIn().setBody("Hello World");
+ getProcessor().process(exchange);
+
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Cannot process", e);
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ Thread thread = new Thread() {
+ @Override
+ public void run() {
+ try {
+ // do not start before the mocks has been setup and is
+ // ready
+ latch.await(5, TimeUnit.SECONDS);
+ doSomething();
+ doSomething();
+ doSomething();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ };
+ thread.start();
+ }
+ }
+}
