diff --git a/build.gradle b/build.gradle index a62de18cbc..1dd8b0b55f 100644 --- a/build.gradle +++ b/build.gradle @@ -54,6 +54,7 @@ ext { assertkVersion = '0.25' avroVersion = '1.11.0' awaitilityVersion = '4.2.0' + camelVersion = '3.18.2' commonsDbcp2Version = '2.9.0' commonsIoVersion = '2.11.0' commonsNetVersion = '3.8.0' @@ -163,6 +164,7 @@ allprojects { mavenBom "org.mockito:mockito-bom:$mockitoVersion" mavenBom "io.micrometer:micrometer-bom:$micrometerVersion" mavenBom "io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion" + mavenBom "org.apache.camel:camel-bom:$camelVersion" } } @@ -477,6 +479,18 @@ project('spring-integration-amqp') { } } +project('spring-integration-camel') { + description = 'Spring Integration support for Apache Camel' + + dependencies { + api project(':spring-integration-core') + api 'org.apache.camel:camel-api' + + testImplementation 'org.apache.camel:camel-test-junit5' + } +} + + project('spring-integration-core') { description = 'Spring Integration Core' diff --git a/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/CamelMessageHandler.java b/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/CamelMessageHandler.java new file mode 100644 index 0000000000..29db655d2e --- /dev/null +++ b/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/CamelMessageHandler.java @@ -0,0 +1,202 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.camel.outbound; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +import org.apache.camel.ProducerTemplate; + +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.camel.support.CamelHeaderMapper; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * A {@link org.springframework.messaging.MessageHandler} for calling Apache Camel route + * and produce (optionally) a reply. + *

+ * In the async mode, the {@link ProducerTemplate#asyncSend(Endpoint, Exchange)} is used. + *

+ * The request-reply behavior can be controlled via {@link ExchangePattern} configuration + * or per message. By default, this handler works in an {@link ExchangePattern#InOnly} mode. + *

+ * A default "mapping all headers" between Spring Integration and Apache Camel messages behavior + * can be customized via {@link #setHeaderMapper(HeaderMapper)} option. + *

+ * The target Apache Camel endpoint to call can be determined by the {@link #endpointUriExpression}. + * By default, a {@link ProducerTemplate#getDefaultEndpoint()} is used. + * + * @author Artem Bilan + * + * @since 6.0 + * + * @see CamelHeaderMapper + */ +public class CamelMessageHandler extends AbstractReplyProducingMessageHandler { + + private final ProducerTemplate producerTemplate; + + private Expression exchangePatternExpression = new ValueExpression<>(ExchangePattern.InOnly); + + @Nullable + private Expression endpointUriExpression; + + private HeaderMapper headerMapper = new CamelHeaderMapper(); + + @Nullable + private Expression exchangePropertiesExpression; + + private StandardEvaluationContext evaluationContext; + + public CamelMessageHandler(ProducerTemplate producerTemplate) { + Assert.notNull(producerTemplate, "'producerTemplate' must not be null"); + this.producerTemplate = producerTemplate; + } + + public void setEndpointUri(String endpointUri) { + Assert.hasText(endpointUri, "'endpointUri' must not be empty"); + setEndpointUriExpression(new LiteralExpression(endpointUri)); + } + + public void setEndpointUriExpression(Expression endpointUriExpression) { + Assert.notNull(endpointUriExpression, "'endpointUriExpression' must not be null"); + this.endpointUriExpression = endpointUriExpression; + } + + public void setExchangePattern(ExchangePattern exchangePattern) { + Assert.notNull(exchangePattern, "'exchangePattern' must not be null"); + setExchangePatternExpression(new ValueExpression<>(exchangePattern)); + } + + public void setExchangePatternExpression(Expression exchangePatternExpression) { + Assert.notNull(exchangePatternExpression, "'exchangePatternExpression' must not be null"); + this.exchangePatternExpression = exchangePatternExpression; + } + + /** + * Set a {@link HeaderMapper} to map request message headers into Apache Camel message headers and + * back if request-reply exchange pattern is used. + * @param headerMapper the {@link HeaderMapper} to use. + */ + public void setHeaderMapper(HeaderMapper headerMapper) { + Assert.notNull(headerMapper, "'headerMapper' must not be null"); + this.headerMapper = headerMapper; + } + + public void setExchangeProperties(Map exchangeProperties) { + Assert.notNull(exchangeProperties, "'exchangeProperties' must not be null"); + setExchangePropertiesExpression(new ValueExpression<>(exchangeProperties)); + } + + /** + * Set a SpEL expression to evaluate {@link org.apache.camel.Exchange} properties as a {@link Map}. + * @param exchangePropertiesExpression the expression for exchange properties. + */ + public void setExchangePropertiesExpression(Expression exchangePropertiesExpression) { + this.exchangePropertiesExpression = exchangePropertiesExpression; + } + + @Override + protected final void doInit() { + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + ExchangePattern exchangePattern = + this.exchangePatternExpression.getValue(this.evaluationContext, requestMessage, ExchangePattern.class); + + Assert.notNull(exchangePattern, "'exchangePatternExpression' must not evaluate to null"); + + Endpoint endpoint = resolveEndpoint(requestMessage); + Exchange exchange = prepareInExchange(endpoint, exchangePattern, requestMessage); + + if (isAsync()) { + CompletableFuture result = this.producerTemplate.asyncSend(endpoint, exchange); + return result.thenApply(resultExchange -> buildReply(exchangePattern, resultExchange)); + } + else { + Exchange result = this.producerTemplate.send(endpoint, exchange); + return buildReply(exchangePattern, result); + } + } + + private Endpoint resolveEndpoint(Message requestMessage) { + String endpointUri = + this.endpointUriExpression != null + ? this.endpointUriExpression.getValue(this.evaluationContext, requestMessage, String.class) + : null; + + if (StringUtils.hasText(endpointUri)) { + return this.producerTemplate.getCamelContext().getEndpoint(endpointUri); + } + else { + return this.producerTemplate.getDefaultEndpoint(); + } + } + + @SuppressWarnings("unchecked") + private Exchange prepareInExchange(Endpoint endpoint, ExchangePattern exchangePattern, Message requestMessage) { + Exchange exchange = endpoint.createExchange(exchangePattern); + + Map exchangeProperties = + this.exchangePropertiesExpression != null + ? this.exchangePropertiesExpression.getValue(this.evaluationContext, requestMessage, Map.class) + : null; + + if (exchangeProperties != null) { + for (Map.Entry property : exchangeProperties.entrySet()) { + exchange.setProperty(property.getKey(), property.getValue()); + } + } + org.apache.camel.Message in = exchange.getIn(); + this.headerMapper.fromHeaders(requestMessage.getHeaders(), in); + in.setBody(requestMessage.getPayload()); + return exchange; + } + + @Nullable + private AbstractIntegrationMessageBuilder buildReply(ExchangePattern exchangePattern, Exchange result) { + if (result.isFailed()) { + throw CamelExecutionException.wrapCamelExecutionException(result, result.getException()); + } + if (exchangePattern.isOutCapable()) { + org.apache.camel.Message out = result.getMessage(); + return getMessageBuilderFactory() + .withPayload(out.getBody()) + .copyHeaders(this.headerMapper.toHeaders(out)); + } + else { + return null; + } + } + +} diff --git a/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/package-info.java b/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/package-info.java new file mode 100644 index 0000000000..7206e2501a --- /dev/null +++ b/spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/package-info.java @@ -0,0 +1,7 @@ +/** + * Provides classes for Apache Camel outbound channel adapters. + */ + +@org.springframework.lang.NonNullApi +@org.springframework.lang.NonNullFields +package org.springframework.integration.camel.outbound; diff --git a/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/CamelHeaderMapper.java b/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/CamelHeaderMapper.java new file mode 100644 index 0000000000..f1d39b28de --- /dev/null +++ b/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/CamelHeaderMapper.java @@ -0,0 +1,115 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.camel.support; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +import org.apache.camel.Message; + +import org.springframework.core.log.LogAccessor; +import org.springframework.core.log.LogMessage; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.Assert; +import org.springframework.util.PatternMatchUtils; + +/** + * A {@link HeaderMapper} for mapping headers from Spring Integration message + * to Apache Camel message and back. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class CamelHeaderMapper implements HeaderMapper { + + + private static final LogAccessor LOGGER = new LogAccessor(CamelHeaderMapper.class); + + private String[] inboundHeaderNames = { "*" }; + + private String[] outboundHeaderNames = { "*" }; + + /** + * Provide a list of patterns to map Apache Camel message headers into Spring Integration message. + * By default, it maps all. + * @param inboundHeaderNames the Apache Camel message headers patterns to map. + */ + public void setInboundHeaderNames(String... inboundHeaderNames) { + Assert.notNull(inboundHeaderNames, "'inboundHeaderNames' must not be null"); + String[] copy = Arrays.copyOf(inboundHeaderNames, inboundHeaderNames.length); + Arrays.sort(copy); + this.inboundHeaderNames = copy; + } + + /** + * Provide a list of patterns to map Spring Integration message headers into an Apache Camel message. + * By default, it maps all. + * @param outboundHeaderNames the header patterns to map. + */ + public void setOutboundHeaderNames(String... outboundHeaderNames) { + Assert.notNull(outboundHeaderNames, "'outboundHeaderNames' must not be null"); + String[] copy = Arrays.copyOf(outboundHeaderNames, outboundHeaderNames.length); + Arrays.sort(copy); + this.outboundHeaderNames = copy; + } + + @Override + public void fromHeaders(MessageHeaders headers, Message target) { + for (Map.Entry entry : headers.entrySet()) { + String name = entry.getKey(); + if (shouldMapHeader(name, this.outboundHeaderNames)) { + Object value = entry.getValue(); + if (value != null) { + target.setHeader(name, value); + } + } + } + } + + @Override + public Map toHeaders(Message source) { + Map headers = new HashMap<>(); + for (Map.Entry entry : source.getHeaders().entrySet()) { + String name = entry.getKey(); + if (shouldMapHeader(name, this.inboundHeaderNames)) { + Object value = entry.getValue(); + if (value != null) { + headers.put(name, value); + } + } + } + return headers; + } + + private static boolean shouldMapHeader(String headerName, String[] patterns) { + if (patterns.length > 0) { + for (String pattern : patterns) { + if (PatternMatchUtils.simpleMatch(pattern, headerName)) { + LOGGER.debug(LogMessage.format("headerName=[{0}] WILL be mapped, matched pattern={1}", + headerName, pattern)); + return true; + } + } + } + LOGGER.debug(LogMessage.format("headerName=[{0}] WILL NOT be mapped", headerName)); + return false; + } + +} diff --git a/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/package-info.java b/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/package-info.java new file mode 100644 index 0000000000..c7a863cc77 --- /dev/null +++ b/spring-integration-camel/src/main/java/org/springframework/integration/camel/support/package-info.java @@ -0,0 +1,7 @@ +/** + * Provides supporting classes for Apache Camel channel adapters. + */ + +@org.springframework.lang.NonNullApi +@org.springframework.lang.NonNullFields +package org.springframework.integration.camel.support; diff --git a/spring-integration-camel/src/test/java/org/springframework/integration/camel/outbound/CamelMessageHandlerTests.java b/spring-integration-camel/src/test/java/org/springframework/integration/camel/outbound/CamelMessageHandlerTests.java new file mode 100644 index 0000000000..94ecb4cffa --- /dev/null +++ b/spring-integration-camel/src/test/java/org/springframework/integration/camel/outbound/CamelMessageHandlerTests.java @@ -0,0 +1,189 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.camel.outbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ExchangePattern; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit5.CamelTestSupport; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.camel.support.CamelHeaderMapper; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Artem Bilan + * + * @since 6.0 + */ +public class CamelMessageHandlerTests extends CamelTestSupport { + + @Test + void inOnlyPatternSyncMessageHandler() throws InterruptedException { + Message messageUnderTest = new GenericMessage<>("Hello Camel!"); + Message messageUnderTest2 = new GenericMessage<>("Hello Camel again!"); + + MockEndpoint mockEndpoint = getMockEndpoint("mock:result"); + mockEndpoint.message(0).body().isEqualTo(messageUnderTest.getPayload()); + mockEndpoint.message(0).header(MessageHeaders.ID).isEqualTo(messageUnderTest.getHeaders().getId()); + mockEndpoint.message(1).body().isEqualTo(messageUnderTest2.getPayload()); + mockEndpoint.message(1).header(MessageHeaders.ID).isEqualTo(messageUnderTest2.getHeaders().getId()); + + CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template()); + camelMessageHandler.setEndpointUri("direct:simple"); + camelMessageHandler.setBeanFactory(mock(BeanFactory.class)); + camelMessageHandler.afterPropertiesSet(); + + camelMessageHandler.handleMessage(messageUnderTest); + camelMessageHandler.handleMessage(messageUnderTest2); + + assertMockEndpointsSatisfied(); + } + + @Test + void inOutPatternSyncMessageHandlerWithNoRequestHeadersButReplyHeaders() throws InterruptedException { + SpelExpressionParser spelExpressionParser = new SpelExpressionParser(); + QueueChannel replyChannel = new QueueChannel(); + Message messageUnderTest = + MessageBuilder.withPayload("test data") + .setHeader("exchangePattern", "InOptionalOut") + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); + + MockEndpoint mockEndpoint = getMockEndpoint("mock:result"); + mockEndpoint.expectedHeaderReceived(MessageHeaders.ID, null); + mockEndpoint.expectedHeaderReceived(MessageHeaders.TIMESTAMP, null); + mockEndpoint.whenAnyExchangeReceived(exchange -> { + org.apache.camel.Message out = exchange.getMessage(); + out.setBody("Reply for: " + exchange.getIn().getBody()); + out.setHeader("testHeader", "testHeaderValue"); + out.setHeader("notMappedHeader", "someValue"); + }); + + CamelHeaderMapper headerMapper = new CamelHeaderMapper(); + headerMapper.setOutboundHeaderNames(""); + headerMapper.setInboundHeaderNames("testHeader"); + + CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template()); + camelMessageHandler.setEndpointUriExpression(new FunctionExpression<>(m -> "direct:simple")); + camelMessageHandler.setExchangePatternExpression(spelExpressionParser.parseExpression("headers.exchangePattern")); + camelMessageHandler.setHeaderMapper(headerMapper); + camelMessageHandler.setBeanFactory(mock(BeanFactory.class)); + camelMessageHandler.afterPropertiesSet(); + + camelMessageHandler.handleMessage(messageUnderTest); + + Message receive = replyChannel.receive(10_000); + + assertMockEndpointsSatisfied(); + + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("Reply for: test data"); + assertThat(receive.getHeaders()) + .containsEntry("testHeader", "testHeaderValue") + .doesNotContainKey("notMappedHeader"); + } + + @Test + void inOnlyPatternAsyncMessageHandlerWithException() throws InterruptedException { + QueueChannel errorChannel = new QueueChannel(); + Message messageUnderTest = + MessageBuilder.withPayload("test data") + .setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel) + .build(); + + getMockEndpoint("mock:result") + .whenAnyExchangeReceived(exchange -> { + throw new RuntimeException("intentional"); + }); + + CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template()); + camelMessageHandler.setEndpointUri("direct:simple"); + camelMessageHandler.setBeanFactory(mock(BeanFactory.class)); + camelMessageHandler.setAsync(true); + camelMessageHandler.afterPropertiesSet(); + + camelMessageHandler.handleMessage(messageUnderTest); + Message receive = errorChannel.receive(10_000); + + assertMockEndpointsSatisfied(); + + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()) + .asInstanceOf(InstanceOfAssertFactories.throwable(MessageHandlingException.class)) + .hasCauseInstanceOf(CamelExecutionException.class) + .hasRootCauseExactlyInstanceOf(RuntimeException.class) + .hasStackTraceContaining("intentional"); + } + + @Test + void inOutPatternAsyncMessageHandler() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message messageUnderTest = + MessageBuilder.withPayload("test async data") + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); + + MockEndpoint mockEndpoint = getMockEndpoint("mock:result"); + mockEndpoint.whenAnyExchangeReceived(exchange -> + exchange.getMessage().setBody("Async reply for: " + exchange.getIn().getBody())); + + ProducerTemplate producerTemplate = template(); + producerTemplate.setDefaultEndpointUri("direct:simple"); + CamelMessageHandler camelMessageHandler = new CamelMessageHandler(producerTemplate); + camelMessageHandler.setExchangePattern(ExchangePattern.InOut); + camelMessageHandler.setBeanFactory(mock(BeanFactory.class)); + camelMessageHandler.setAsync(true); + camelMessageHandler.afterPropertiesSet(); + + camelMessageHandler.handleMessage(messageUnderTest); + + Message receive = replyChannel.receive(10_000); + + assertMockEndpointsSatisfied(); + + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("Async reply for: test async data"); + } + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + + @Override + public void configure() { + from("direct:simple").to("mock:result"); + } + }; + } + +} diff --git a/spring-integration-camel/src/test/resources/log4j2-test.xml b/spring-integration-camel/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000000..32403b74b5 --- /dev/null +++ b/spring-integration-camel/src/test/resources/log4j2-test.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 4a2989e1a6..f7e0bffc0d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @@ -535,7 +536,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan @Override public void accept(Object result, Throwable exception) { - if (exception == null) { + if (result != null) { Message replyMessage = null; try { replyMessage = createOutputMessage(result, this.requestMessage.getHeaders()); @@ -549,12 +550,12 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); } } - logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result.toString()); + logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result); onFailure(exceptionToLogAndSend); } } - else { - onFailure(exception); + else if (exception != null) { + onFailure(exception instanceof CompletionException ? exception.getCause() : exception); } }