From 7f74c571d47cfab8aac159fe81db5cc963a55d4c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 21 Aug 2014 11:16:01 +0300 Subject: [PATCH] INT-3402-3: Channels Late Resolution #3 JIRA: https://jira.spring.io/browse/INT-3402 * Add late resolution of channel names for the `MessagingGatewaySupport` * Implement delegate logic for internal implementations like `ContentEnricher.Gateway` --- .../gateway/MessagingGatewaySupport.java | 167 +++++++++++++++--- .../transformer/ContentEnricher.java | 37 ++-- .../xml/EnricherParserTests-context.xml | 2 +- .../config/xml/EnricherParserTests.java | 19 +- .../transformer/ContentEnricherTests.java | 45 +++-- .../ChannelPublishingJmsMessageListener.java | 69 ++++---- 6 files changed, 237 insertions(+), 102 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 4047b76de5..9d7b454a28 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -16,6 +16,7 @@ package org.springframework.integration.gateway; +import org.springframework.beans.BeansException; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; @@ -33,6 +34,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; @@ -53,10 +55,16 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private volatile MessageChannel requestChannel; + private volatile String requestChannelName; + private volatile MessageChannel replyChannel; + private volatile String replyChannelName; + private volatile MessageChannel errorChannel; + private volatile String errorChannelName; + private volatile long replyTimeout = DEFAULT_TIMEOUT; @SuppressWarnings("rawtypes") @@ -66,7 +74,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private final MessagingTemplate messagingTemplate; - private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); + private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = + new HistoryWritingMessagePostProcessor(); private volatile boolean initialized; @@ -86,38 +95,67 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the request channel. - * * @param requestChannel the channel to which request messages will be sent */ public void setRequestChannel(MessageChannel requestChannel) { this.requestChannel = requestChannel; } + /** + * Set the request channel name. + * @param requestChannelName the channel bean name to which request messages will be sent + * @since 4.1 + */ + public void setRequestChannelName(String requestChannelName) { + Assert.hasText(requestChannelName, "'requestChannelName' must not be empty"); + this.requestChannelName = requestChannelName; + } + /** * Set the reply channel. If no reply channel is provided, this gateway will * always use an anonymous, temporary channel for handling replies. - * * @param replyChannel the channel from which reply messages will be received */ public void setReplyChannel(MessageChannel replyChannel) { this.replyChannel = replyChannel; } + /** + * Set the reply channel name. If no reply channel is provided, this gateway will + * always use an anonymous, temporary channel for handling replies. + * @param replyChannelName the channel bean name from which reply messages will be received + * @since 4.1 + */ + public void setReplyChannelName(String replyChannelName) { + Assert.hasText(replyChannelName, "'replyChannelName' must not be empty"); + this.replyChannelName = replyChannelName; + } + /** * Set the error channel. If no error channel is provided, this gateway will * propagate Exceptions to the caller. To completely suppress Exceptions, provide * a reference to the "nullChannel" here. - * * @param errorChannel The error channel. */ public void setErrorChannel(MessageChannel errorChannel) { this.errorChannel = errorChannel; } + /** + * Set the error channel name. If no error channel is provided, this gateway will + * propagate Exceptions to the caller. To completely suppress Exceptions, provide + * a reference to the "nullChannel" here. + * @param errorChannelName The error channel bean name. + * @since 4.1 + */ + public void setErrorChannelName(String errorChannelName) { + Assert.hasText(errorChannelName, "'errorChannelName' must not be empty"); + this.errorChannelName = errorChannelName; + } + /** * Set the timeout value for sending request messages. If not * explicitly configured, the default is one second. - * * @param requestTimeout the timeout value in milliseconds */ public void setRequestTimeout(long requestTimeout) { @@ -127,7 +165,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the timeout value for receiving reply messages. If not * explicitly configured, the default is one second. - * * @param replyTimeout the timeout value in milliseconds */ public void setReplyTimeout(long replyTimeout) { @@ -138,7 +175,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Provide an {@link InboundMessageMapper} for creating request Messages * from any object passed in a send or sendAndReceive operation. - * * @param requestMapper The request mapper. */ public void setRequestMapper(InboundMessageMapper requestMapper) { @@ -150,7 +186,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Provide an {@link OutboundMessageMapper} for mapping to objects from * any reply Messages received in receive or sendAndReceive operations. - * * @param replyMapper The reply mapper. */ public void setReplyMapper(OutboundMessageMapper replyMapper) { @@ -173,6 +208,12 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement @Override protected void onInit() throws Exception { + Assert.state(!(this.requestChannelName != null && this.requestChannel != null), + "'requestChannelName' and 'requestChannel' are mutually exclusive."); + Assert.state(!(this.replyChannelName != null && this.replyChannel != null), + "'replyChannelName' and 'replyChannel' are mutually exclusive."); + Assert.state(!(this.errorChannelName != null && this.errorChannel != null), + "'errorChannelName' and 'errorChannel' are mutually exclusive."); this.historyWritingPostProcessor.setTrackableComponent(this); this.historyWritingPostProcessor.setMessageBuilderFactory(this.getMessageBuilderFactory()); if (this.getBeanFactory() != null) { @@ -191,17 +232,79 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement } } + private MessageChannel getRequestChannel() { + if (this.requestChannelName != null) { + synchronized (this) { + if (this.requestChannelName != null) { + try { + Assert.state(getBeanFactory() != null, + "A bean factory is required to resolve the requestChannel at runtime."); + this.requestChannel = getBeanFactory().getBean(this.requestChannelName, MessageChannel.class); + this.requestChannelName = null; + } + catch (BeansException e) { + throw new DestinationResolutionException("Failed to look up MessageChannel with name '" + + this.requestChannelName + "' in the BeanFactory."); + } + } + } + } + return this.requestChannel; + } + + private MessageChannel getReplyChannel() { + if (this.replyChannelName != null) { + synchronized (this) { + if (this.replyChannelName != null) { + try { + Assert.state(getBeanFactory() != null, + "A bean factory is required to resolve the replyChannel at runtime."); + this.replyChannel = getBeanFactory().getBean(this.replyChannelName, MessageChannel.class); + this.replyChannelName = null; + } + catch (BeansException e) { + throw new DestinationResolutionException("Failed to look up MessageChannel with name '" + + this.replyChannelName + "' in the BeanFactory."); + } + } + } + } + return this.replyChannel; + } + + private MessageChannel getErrorChannel() { + if (this.errorChannelName != null) { + synchronized (this) { + if (this.errorChannelName != null) { + try { + Assert.state(getBeanFactory() != null, + "A bean factory is required to resolve the errorChannel at runtime."); + this.errorChannel = getBeanFactory().getBean(this.errorChannelName, MessageChannel.class); + this.errorChannelName = null; + } + catch (BeansException e) { + throw new DestinationResolutionException("Failed to look up MessageChannel with name '" + + this.errorChannelName + "' in the BeanFactory."); + } + } + } + } + return this.errorChannel; + } + protected void send(Object object) { this.initializeIfNecessary(); Assert.notNull(object, "request must not be null"); - Assert.state(this.requestChannel != null, + MessageChannel requestChannel = getRequestChannel(); + Assert.state(requestChannel != null, "send is not supported, because no request channel has been configured"); try { - this.messagingTemplate.convertAndSend(this.requestChannel, object, this.historyWritingPostProcessor); + this.messagingTemplate.convertAndSend(requestChannel, object, this.historyWritingPostProcessor); } catch (Exception e) { - if (this.errorChannel != null) { - this.messagingTemplate.send(this.errorChannel, new ErrorMessage(e)); + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { + this.messagingTemplate.send(errorChannel, new ErrorMessage(e)); } else { this.rethrow(e, "failed to send message"); @@ -211,9 +314,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement protected Object receive() { this.initializeIfNecessary(); - Assert.state(this.replyChannel != null && (this.replyChannel instanceof PollableChannel), + MessageChannel replyChannel = getReplyChannel(); + Assert.state(replyChannel != null && (replyChannel instanceof PollableChannel), "receive is not supported, because no pollable reply channel has been configured"); - return this.messagingTemplate.receiveAndConvert((PollableChannel) this.replyChannel, null); + return this.messagingTemplate.receiveAndConvert(replyChannel, null); } protected Object sendAndReceive(Object object) { @@ -228,17 +332,20 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private Object doSendAndReceive(Object object, boolean shouldConvert) { this.initializeIfNecessary(); Assert.notNull(object, "request must not be null"); - if (this.requestChannel == null) { + MessageChannel requestChannel = getRequestChannel(); + if (requestChannel == null) { throw new MessagingException("No request channel available. Cannot send request message."); } - if (this.replyChannel != null && this.replyMessageCorrelator == null) { + MessageChannel replyChannel = getReplyChannel(); + if (replyChannel != null && this.replyMessageCorrelator == null) { this.registerReplyMessageCorrelator(); } Object reply = null; Throwable error = null; try { if (shouldConvert) { - reply = this.messagingTemplate.convertSendAndReceive(this.requestChannel, object, null, this.historyWritingPostProcessor); + reply = this.messagingTemplate.convertSendAndReceive(requestChannel, object, null, + this.historyWritingPostProcessor); if (reply instanceof Throwable) { error = (Throwable) reply; } @@ -247,7 +354,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement Message requestMessage = (object instanceof Message) ? (Message) object : this.requestMapper.toMessage(object); requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage); - reply = this.messagingTemplate.sendAndReceive(this.requestChannel, requestMessage); + reply = this.messagingTemplate.sendAndReceive(requestChannel, requestMessage); if (reply instanceof ErrorMessage) { error = ((ErrorMessage) reply).getPayload(); } @@ -261,14 +368,16 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement } if (error != null) { - if (this.errorChannel != null) { + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { Message errorMessage = new ErrorMessage(error); Message errorFlowReply = null; try { - errorFlowReply = this.messagingTemplate.sendAndReceive(this.errorChannel, errorMessage); + errorFlowReply = this.messagingTemplate.sendAndReceive(errorChannel, errorMessage); } catch (Exception errorFlowFailure) { - throw new MessagingException(errorMessage, "failure occurred in error-handling flow", errorFlowFailure); + throw new MessagingException(errorMessage, "failure occurred in error-handling flow", + errorFlowFailure); } if (shouldConvert) { Object result = (errorFlowReply != null) ? errorFlowReply.getPayload() : null; @@ -307,18 +416,21 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement handler.setBeanFactory(this.getBeanFactory()); } handler.afterPropertiesSet(); - if (this.replyChannel instanceof SubscribableChannel) { - correlator = new EventDrivenConsumer( - (SubscribableChannel) this.replyChannel, handler); + MessageChannel replyChannel = getReplyChannel(); + if (replyChannel instanceof SubscribableChannel) { + correlator = new EventDrivenConsumer((SubscribableChannel) replyChannel, handler); } - else if (this.replyChannel instanceof PollableChannel) { - PollingConsumer endpoint = new PollingConsumer( - (PollableChannel) this.replyChannel, handler); + else if (replyChannel instanceof PollableChannel) { + PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChannel, handler); endpoint.setBeanFactory(this.getBeanFactory()); endpoint.setReceiveTimeout(this.replyTimeout); endpoint.afterPropertiesSet(); correlator = endpoint; } + else { + throw new MessagingException("Unsupported 'replyChannel' type [" + replyChannel.getClass() + "]." + + "SubscribableChannel or PollableChannel type are supported."); + } if (this.isRunning()) { correlator.start(); } @@ -356,6 +468,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement } return (object != null) ? this.messageBuilderFactory.withPayload(object).build() : null; } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index 946d477aa4..bab53b0dd8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -37,7 +37,6 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; -import org.springframework.util.StringUtils; /** * Content Enricher is a Message Transformer that can augment a message's payload with @@ -55,6 +54,8 @@ import org.springframework.util.StringUtils; public class ContentEnricher extends AbstractReplyProducingMessageHandler implements Lifecycle, IntegrationEvaluationContextAware { + private final SpelExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true)); + private volatile Map nullResultPropertyExpressions = new HashMap(); private volatile Map> nullResultHeaderExpressions = @@ -65,8 +66,6 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler private volatile Map> headerExpressions = new HashMap>(); - private final SpelExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true)); - private EvaluationContext sourceEvaluationContext; private EvaluationContext targetEvaluationContext; @@ -149,6 +148,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler } public void setRequestChannelName(String requestChannelName) { + Assert.hasText(requestChannelName, "'requestChannelName' must not be empty"); this.requestChannelName = requestChannelName; } @@ -163,6 +163,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler } public void setReplyChannelName(String replyChannelName) { + Assert.hasText(replyChannelName, "'replyChannelName' must not be empty"); this.replyChannelName = replyChannelName; } @@ -239,33 +240,33 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler */ @Override protected void doInit() { - if (StringUtils.hasText(this.requestChannelName)) { - Assert.isNull(this.requestChannel, "'requestChannelName' and 'requestChannel' are mutually exclusive."); - this.requestChannel = this.getBeanFactory().getBean(this.requestChannelName, MessageChannel.class); - } + Assert.state(!(this.requestChannelName != null && this.requestChannel != null), + "'requestChannelName' and 'requestChannel' are mutually exclusive."); - if (StringUtils.hasText(this.replyChannelName)) { - Assert.isNull(this.replyChannel, "'replyChannelName' and 'replyChannel' are mutually exclusive."); - this.replyChannel = this.getBeanFactory().getBean(this.replyChannelName, MessageChannel.class); - } + Assert.state(!(this.replyChannelName != null && this.replyChannel != null), + "'replyChannelName' and 'replyChannel' are mutually exclusive."); - if (this.replyChannel != null) { - Assert.notNull(this.requestChannel, "If the replyChannel is set, then the requestChannel must not be null"); + if (this.replyChannel != null || this.replyChannelName != null) { + Assert.state(this.requestChannel != null || this.requestChannelName != null, + "If the replyChannel is set, then the requestChannel must not be null"); } - if (this.requestChannel != null) { + if (this.requestChannel != null || this.requestChannelName != null) { this.gateway = new Gateway(); - this.gateway.setRequestChannel(requestChannel); + this.gateway.setRequestChannel(this.requestChannel); + if (this.requestChannelName != null) { + this.gateway.setRequestChannelName(this.requestChannelName); + } if (this.requestTimeout != null) { this.gateway.setRequestTimeout(this.requestTimeout); } - if (this.replyTimeout != null) { this.gateway.setReplyTimeout(this.replyTimeout); } - if (replyChannel != null) { - this.gateway.setReplyChannel(replyChannel); + this.gateway.setReplyChannel(replyChannel); + if (this.replyChannelName != null) { + this.gateway.setReplyChannelName(this.replyChannelName); } if (this.getBeanFactory() != null) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml index 73eb705c68..09b275f59b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml @@ -44,7 +44,7 @@ -
+
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java index d693b3b349..7709143796 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -34,6 +34,7 @@ import org.springframework.beans.TypeMismatchException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.expression.Expression; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; @@ -44,7 +45,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; @@ -82,7 +82,8 @@ public class EnricherParserTests { assertNull(accessor.getPropertyValue("requestPayloadExpression")); assertNotNull(TestUtils.getPropertyValue(enricher, "gateway.beanFactory")); - Map propertyExpressions = (Map) accessor.getPropertyValue("propertyExpressions"); + Map propertyExpressions = + (Map) accessor.getPropertyValue("propertyExpressions"); for (Map.Entry e : propertyExpressions.entrySet()) { if ("name".equals(e.getKey().getExpressionString())) { assertEquals("payload.sourceName", e.getValue().getExpressionString()); @@ -110,7 +111,7 @@ public class EnricherParserTests { Object endpoint = context.getBean("enricher"); Long requestTimeout = TestUtils.getPropertyValue(endpoint, "handler.requestTimeout", Long.class); - Long replyTimeout = TestUtils.getPropertyValue(endpoint, "handler.replyTimeout", Long.class); + Long replyTimeout = TestUtils.getPropertyValue(endpoint, "handler.replyTimeout", Long.class); assertEquals(Long.valueOf(1234L), requestTimeout); assertEquals(Long.valueOf(9876L), replyTimeout); @@ -130,6 +131,9 @@ public class EnricherParserTests { @Test public void integrationTest() { + QueueChannel output = context.getBean("output", QueueChannel.class); + output.purge(null); + SubscribableChannel requests = context.getBean("requests", SubscribableChannel.class); class Foo extends AbstractReplyProducingMessageHandler { @@ -149,7 +153,8 @@ public class EnricherParserTests { .setHeader("notOverwrite", "test") .build(); context.getBean("input", MessageChannel.class).send(request); - Message reply = context.getBean("output", PollableChannel.class).receive(0); + + Message reply = output.receive(0); Target enriched = (Target) reply.getPayload(); assertEquals("foo", enriched.getName()); assertEquals(42, enriched.getAge()); @@ -193,6 +198,7 @@ public class EnricherParserTests { public String getSourceName() { return sourceName; } + } public static class Target implements Cloneable { @@ -246,10 +252,12 @@ public class EnricherParserTests { copy.setMarried(this.married); return copy; } + } public static enum Gender { MALE, FEMALE + } public static class FooAdvice extends AbstractRequestHandlerAdvice { @@ -261,4 +269,5 @@ public class EnricherParserTests { } } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java index 7a387e8934..e63228c663 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java @@ -118,7 +118,8 @@ public class ContentEnricherTests { protected Object handleRequestMessage(Message requestMessage) { try { Thread.sleep(5000); - } catch (InterruptedException e) { + } + catch (InterruptedException e) { fail(e.getMessage()); } return new Target("child"); @@ -144,7 +145,8 @@ public class ContentEnricherTests { try { enricher.handleMessage(requestMessage); - } catch (ReplyRequiredException e) { + } + catch (ReplyRequiredException e) { assertEquals("No reply produced by handler 'Enricher', and its 'requiresReply' property is set to true.", e.getMessage()); return; } @@ -173,8 +175,9 @@ public class ContentEnricherTests { Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); try { - enricher.handleMessage(requestMessage); - } catch (MessageDeliveryException e) { + enricher.handleMessage(requestMessage); + } + catch (MessageDeliveryException e) { assertEquals("failed to send message to channel '" + requestChannelName + "' within timeout: " + requestTimeout, e.getMessage()); return; @@ -220,8 +223,9 @@ public class ContentEnricherTests { enricher.setBeanFactory(mock(BeanFactory.class)); try { - enricher.afterPropertiesSet(); - } catch (IllegalArgumentException e) { + enricher.afterPropertiesSet(); + } + catch (IllegalStateException e) { assertEquals("If the replyChannel is set, then the requestChannel must not be null", e.getMessage()); return; } @@ -236,8 +240,9 @@ public class ContentEnricherTests { enricher.setBeanFactory(mock(BeanFactory.class)); try { - enricher.setReplyTimeout(null); - } catch (IllegalArgumentException e) { + enricher.setReplyTimeout(null); + } + catch (IllegalArgumentException e) { assertEquals("replyTimeout must not be null", e.getMessage()); return; } @@ -252,8 +257,9 @@ public class ContentEnricherTests { enricher.setBeanFactory(mock(BeanFactory.class)); try { - enricher.setRequestTimeout(null); - } catch (IllegalArgumentException e) { + enricher.setRequestTimeout(null); + } + catch (IllegalArgumentException e) { assertEquals("requestTimeout must not be null", e.getMessage()); return; } @@ -286,10 +292,11 @@ public class ContentEnricherTests { enricher.setBeanFactory(mock(BeanFactory.class)); try { - enricher.afterPropertiesSet(); - } catch (IllegalArgumentException e) { - assertEquals("If the replyChannel is set, then the requestChannel must not be null", e.getMessage()); - return; + enricher.afterPropertiesSet(); + } + catch (IllegalStateException e) { + assertEquals("If the replyChannel is set, then the requestChannel must not be null", e.getMessage()); + return; } fail("Expected an IllegalArgumentException to be thrown."); @@ -414,10 +421,11 @@ public class ContentEnricherTests { Message requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build(); try { - enricher.handleMessage(requestMessage); - } catch (MessageHandlingException e) { + enricher.handleMessage(requestMessage); + } + catch (MessageHandlingException e) { assertThat(e.getMessage(), containsString("Failed to clone payload object")); - return; + return; } fail("Expected a MessageHandlingException to be thrown."); @@ -478,6 +486,7 @@ public class ContentEnricherTests { public String getLastName() { return lastName; } + } @@ -517,6 +526,7 @@ public class ContentEnricherTests { clone.setChild(this.child); return clone; } + } public static final class TargetUser { @@ -557,6 +567,7 @@ public class ContentEnricherTests { public Object clone() { throw new IllegalStateException("Cloning not possible"); } + } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 8820b98914..7a3ac3662e 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -56,9 +56,11 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Juergen Hoeller * @author Oleg Zhurakousky + * @author Artem Bilan */ public class ChannelPublishingJmsMessageListener - implements SessionAwareMessageListener, InitializingBean, TrackableComponent, BeanFactoryAware { + implements SessionAwareMessageListener, InitializingBean, + TrackableComponent, BeanFactoryAware { protected final Log logger = LogFactory.getLog(getClass()); @@ -94,34 +96,45 @@ public class ChannelPublishingJmsMessageListener /** * Specify whether a JMS reply Message is expected. - * * @param expectReply true if a reply is expected. */ public void setExpectReply(boolean expectReply) { this.expectReply = expectReply; } - public void setComponentName(String componentName){ + public void setComponentName(String componentName) { this.gatewayDelegate.setComponentName(componentName); } - public void setRequestChannel(MessageChannel requestChannel){ + public void setRequestChannel(MessageChannel requestChannel) { this.gatewayDelegate.setRequestChannel(requestChannel); } - public void setReplyChannel(MessageChannel replyChannel){ + public void setRequestChannelName(String requestChannelName) { + gatewayDelegate.setRequestChannelName(requestChannelName); + } + + public void setReplyChannel(MessageChannel replyChannel) { this.gatewayDelegate.setReplyChannel(replyChannel); } - public void setErrorChannel(MessageChannel errorChannel){ + public void setReplyChannelName(String replyChannelName) { + gatewayDelegate.setReplyChannelName(replyChannelName); + } + + public void setErrorChannel(MessageChannel errorChannel) { this.gatewayDelegate.setErrorChannel(errorChannel); } - public void setRequestTimeout(long requestTimeout){ + public void setErrorChannelName(String errorChannelName) { + gatewayDelegate.setErrorChannelName(errorChannelName); + } + + public void setRequestTimeout(long requestTimeout) { this.gatewayDelegate.setRequestTimeout(requestTimeout); } - public void setReplyTimeout(long replyTimeout){ + public void setReplyTimeout(long replyTimeout) { this.gatewayDelegate.setReplyTimeout(replyTimeout); } @@ -144,7 +157,6 @@ public class ChannelPublishingJmsMessageListener * Set the default reply destination to send reply messages to. This will * be applied in case of a request message that does not carry a * "JMSReplyTo" field. - * * @param defaultReplyDestination The default reply destination. */ public void setDefaultReplyDestination(Destination defaultReplyDestination) { @@ -156,9 +168,7 @@ public class ChannelPublishingJmsMessageListener * This will be applied in case of a request message that does not carry a * "JMSReplyTo" field. *

Alternatively, specify a JMS Destination object as "defaultReplyDestination". - * * @param destinationName The default reply destination name. - * * @see #setDestinationResolver * @see #setDefaultReplyDestination(javax.jms.Destination) */ @@ -171,9 +181,7 @@ public class ChannelPublishingJmsMessageListener * This will be applied in case of a request message that does not carry a * "JMSReplyTo" field. *

Alternatively, specify a JMS Destination object as "defaultReplyDestination". - * * @param destinationName The default reply topic name. - * * @see #setDestinationResolver * @see #setDefaultReplyDestination(javax.jms.Destination) */ @@ -183,9 +191,7 @@ public class ChannelPublishingJmsMessageListener /** * Specify the time-to-live property for JMS reply Messages. - * * @param replyTimeToLive The reply time to live. - * * @see javax.jms.MessageProducer#setTimeToLive(long) */ public void setReplyTimeToLive(long replyTimeToLive) { @@ -194,9 +200,7 @@ public class ChannelPublishingJmsMessageListener /** * Specify the priority value for JMS reply Messages. - * * @param replyPriority The reply priority. - * * @see javax.jms.MessageProducer#setPriority(int) */ public void setReplyPriority(int replyPriority) { @@ -205,9 +209,7 @@ public class ChannelPublishingJmsMessageListener /** * Specify the delivery mode for JMS reply Messages. - * * @param replyDeliveryPersistent true for a persistent reply message. - * * @see javax.jms.MessageProducer#setDeliveryMode(int) */ public void setReplyDeliveryPersistent(boolean replyDeliveryPersistent) { @@ -224,7 +226,6 @@ public class ChannelPublishingJmsMessageListener * instead, then this value should be set to "JMSCorrelationID". * Any other value will be treated as a JMS String Property to be copied as-is * from the request Message into the reply Message with the same property name. - * * @param correlationKey The correlation key. */ public void setCorrelationKey(String correlationKey) { @@ -234,7 +235,6 @@ public class ChannelPublishingJmsMessageListener /** * Specify whether explicit QoS should be enabled for replies * (for timeToLive, priority, and deliveryMode settings). - * * @param explicitQosEnabledForReplies true to enable explicit QoS. */ public void setExplicitQosEnabledForReplies(boolean explicitQosEnabledForReplies) { @@ -246,9 +246,7 @@ public class ChannelPublishingJmsMessageListener * destination names for this listener. *

The default resolver is a DynamicDestinationResolver. Specify a * JndiDestinationResolver for resolving destination names as JNDI locations. - * * @param destinationResolver The destination resolver. - * * @see org.springframework.jms.support.destination.DynamicDestinationResolver * @see org.springframework.jms.support.destination.JndiDestinationResolver */ @@ -262,7 +260,6 @@ public class ChannelPublishingJmsMessageListener * converting between JMS Messages and Spring Integration Messages. * If none is provided, a {@link SimpleMessageConverter} will * be used. - * * @param messageConverter The message converter. */ public void setMessageConverter(MessageConverter messageConverter) { @@ -273,7 +270,6 @@ public class ChannelPublishingJmsMessageListener * Provide a {@link JmsHeaderMapper} implementation to use when * converting between JMS Messages and Spring Integration Messages. * If none is provided, a {@link DefaultJmsHeaderMapper} will be used. - * * @param headerMapper The header mapper. */ public void setHeaderMapper(JmsHeaderMapper headerMapper) { @@ -285,7 +281,6 @@ public class ChannelPublishingJmsMessageListener * to converting into a Spring Integration Message. This value is set to * true by default. To send the JMS Message itself as a * Spring Integration Message payload, set this to false. - * * @param extractRequestPayload true if the request payload should be extracted. */ public void setExtractRequestPayload(boolean extractRequestPayload) { @@ -297,7 +292,6 @@ public class ChannelPublishingJmsMessageListener * extracted prior to converting into a JMS Message. This value is set to * true by default. To send the Spring Integration Message * itself as the JMS Message's body, set this to false. - * * @param extractReplyPayload true if the reply payload should be extracted. */ public void setExtractReplyPayload(boolean extractReplyPayload) { @@ -315,7 +309,8 @@ public class ChannelPublishingJmsMessageListener if (this.extractRequestPayload) { result = this.messageConverter.fromMessage(jmsMessage); if (logger.isDebugEnabled()) { - logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload [" + result + "]"); + logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload [" + + result + "]"); } } @@ -356,7 +351,7 @@ public class ChannelPublishingJmsMessageListener } @Override - public void afterPropertiesSet() { + public void afterPropertiesSet() { if (this.beanFactory != null) { this.gatewayDelegate.setBeanFactory(this.beanFactory); } @@ -364,15 +359,16 @@ public class ChannelPublishingJmsMessageListener this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory); } - protected void start(){ + protected void start() { this.gatewayDelegate.start(); } - protected void stop(){ + protected void stop() { this.gatewayDelegate.stop(); } - private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage) throws JMSException { + private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage) + throws JMSException { if (this.correlationKey != null) { if (this.correlationKey.equals("JMSCorrelationID")) { replyMessage.setJMSCorrelationID(requestMessage.getJMSCorrelationID()); @@ -383,7 +379,8 @@ public class ChannelPublishingJmsMessageListener replyMessage.setStringProperty(this.correlationKey, value); } else if (logger.isWarnEnabled()) { - logger.warn("No property value available on request Message for correlationKey '" + this.correlationKey + "'"); + logger.warn("No property value available on request Message for correlationKey '" + + this.correlationKey + "'"); } } } @@ -442,7 +439,8 @@ public class ChannelPublishingJmsMessageListener return null; } - private void sendReply(javax.jms.Message replyMessage, Destination destination, Session session) throws JMSException { + private void sendReply(javax.jms.Message replyMessage, Destination destination, Session session) + throws JMSException { MessageProducer producer = session.createProducer(destination); try { if (this.explicitQosEnabledForReplies) { @@ -471,6 +469,7 @@ public class ChannelPublishingJmsMessageListener this.name = name; this.isTopic = isTopic; } + } private class GatewayDelegate extends MessagingGatewaySupport { @@ -494,5 +493,7 @@ public class ChannelPublishingJmsMessageListener return "jms:message-driven-channel-adapter"; } } + } + }