Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2748: Add bean definition info into exceptions (#2986)

Browse Source 
* GH-2748: Add bean definition info into exceptions

Fixes https://github.com/spring-projects/spring-integration/issues/2748

In many cases Spring Integration stack traces doesn't contain any
relations to end-user code.
Just because a target project code mostly contains only a configuration
for out-of-the-box components without any custom code.
When exception is thrown from such an out-of-the-box component, it is
hard from the stack trace to determine a configuration source for those
components.

* Add a logic into the `IntegrationObjectSupport` to obtain a its own
`BeanDefinition` from the `BeanFactory` to include a `resource` and
`source` (if any) into the `toString()` representation, as well as add
a new `getBeanDescription()` to get such an info at runtime
* The `toString()` is simply used by `this` reference in the message
for `MessagingException` thrown from the `IntegrationObjectSupport`
implementations
* Modify an exception message for the `MessageTransformingHandler` and
`MessageFilter` to make it based on `this`.
The `AbstractMessageHandler` already includes `this` into its exception
message
* Modify a `AbstractConsumerEndpointParser` and
`AbstractAmqpInboundAdapterParser` (as a sample) to include a `resource`
and `source` into a `MessageHandler` `BeanDefinition`.
* Include an `IntegrationFlow` `BeanDefinition` `resource`
(`@Configuration` class) and its bean method as a `source` into all
child beans declared during flow parsing in the `IntegrationFlowBeanPostProcessor`
* Add `IntegrationFlowRegistrationBuilder.setSource()` for manually
registered flows: there is no configuration parsing phase to extract
such an info from `BeanFactory`
* Propagate that `source` into all the child beans provided by the
`IntegrationFlow`
* Modify a `LambdaMessageProcessor` exception message to include a
method info in case of `InvocationTargetException`

* Do not cast explicitly for `ConfigurableListableBeanFactory` in the
`IntegrationObjectSupport` to avoid tests modifications for mocking
directly into `ConfigurableListableBeanFactory`.
Use `instanceof` instead in the `getBeanDescription()`

* * Fix Checkstyle issues

* * Fix `IntegrationGraphServer` and  `IntegrationMBeanExporter`
to rely on the `NamedComponent` for channel names instead of
always call `toString()` which is now much more than just a bean name
* Don't describe a `componentName` if it is the same as a `beanName`
* Check for parent `BeanDefinition` in the `IntegrationFlowBeanPostProcessor`
before calling its meta-info
* Fix tests according new `IntegrationObjectSupport.toString()` behavior
This commit is contained in:
Artem Bilan
2019-07-17 15:11:10 -04:00
committed by Gary Russell
parent ebb22c2ed4
commit c712416b63
25 changed files with 330 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java
View File

@@ -25,6 +25,7 @@ import org.w3c.dom.Element;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
@@ -39,6 +40,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
@@ -103,7 +105,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
builder.addConstructorArgReference(listenerContainerRef);
}
else {
BeanDefinition listenerContainerBeanDef = this.buildListenerContainer(element, parserContext);
BeanDefinition listenerContainerBeanDef = buildListenerContainer(element, parserContext);
builder.addConstructorArgValue(listenerContainerBeanDef);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
@@ -117,11 +119,12 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
this.configureChannels(element, parserContext, builder);
configureChannels(element, parserContext, builder);
AbstractBeanDefinition adapterBeanDefinition = builder.getRawBeanDefinition();
adapterBeanDefinition.setResource(parserContext.getReaderContext().getResource());
adapterBeanDefinition.setSource(IntegrationNamespaceUtils.createElementDescription(element));
}
protected abstract void configureChannels(Element element, ParserContext parserContext, BeanDefinitionBuilder builder);
private BeanDefinition buildListenerContainer(Element element, ParserContext parserContext) {
if (!element.hasAttribute("queue-names")) {
parserContext.getReaderContext().error("If no 'listener-container' reference is provided, " +
@@ -163,7 +166,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
private void assertNoContainerAttributes(Element element, ParserContext parserContext) {
Object source = parserContext.extractSource(element);
List<String> allContainerAttributes = new ArrayList<String>(Arrays.asList(CONTAINER_VALUE_ATTRIBUTES));
List<String> allContainerAttributes = new ArrayList<>(Arrays.asList(CONTAINER_VALUE_ATTRIBUTES));
allContainerAttributes.addAll(Arrays.asList(CONTAINER_REFERENCE_ATTRIBUTES));
for (String attributeName : allContainerAttributes) {
if (StringUtils.hasText(element.getAttribute(attributeName))) {
@@ -173,4 +176,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
}
}
protected abstract void configureChannels(Element element, ParserContext parserContext,
BeanDefinitionBuilder builder);
}

21
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java
View File

@@ -58,6 +58,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String id = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(id)) {
id = element.getAttribute("name");
@@ -103,10 +104,12 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
}
AbstractBeanDefinition handlerBeanDefinition = handlerBuilder.getBeanDefinition();
String inputChannelAttributeName = this.getInputChannelAttributeName();
handlerBeanDefinition.setResource(parserContext.getReaderContext().getResource());
String elementDescription = IntegrationNamespaceUtils.createElementDescription(element);
handlerBeanDefinition.setSource(elementDescription);
String inputChannelAttributeName = getInputChannelAttributeName();
boolean hasInputChannelAttribute = element.hasAttribute(inputChannelAttributeName);
if (parserContext.isNested()) {
String elementDescription = IntegrationNamespaceUtils.createElementDescription(element);
if (hasInputChannelAttribute) {
parserContext.getReaderContext().error("The '" + inputChannelAttributeName
+ "' attribute isn't allowed for a nested (e.g. inside a <chain/>) endpoint element: "
@@ -122,7 +125,6 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
}
else {
if (!hasInputChannelAttribute) {
String elementDescription = IntegrationNamespaceUtils.createElementDescription(element);
parserContext.getReaderContext().error("The '" + inputChannelAttributeName
+ "' attribute is required for the top-level endpoint element: "
+ elementDescription + ".", element);
@@ -135,9 +137,11 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
builder.addPropertyValue("adviceChain", adviceChain);
}
String handlerBeanName = BeanDefinitionReaderUtils.generateBeanName(handlerBeanDefinition, parserContext.getRegistry());
String handlerBeanName =
BeanDefinitionReaderUtils.generateBeanName(handlerBeanDefinition, parserContext.getRegistry());
String[] handlerAlias = IntegrationNamespaceUtils.generateAlias(element);
parserContext.registerBeanComponent(new BeanComponentDefinition(handlerBeanDefinition, handlerBeanName, handlerAlias));
parserContext.registerBeanComponent(
new BeanComponentDefinition(handlerBeanDefinition, handlerBeanName, handlerAlias));
builder.addPropertyReference("handler", handlerBeanName);
@@ -157,7 +161,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.ROLE);
AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
String beanName = this.resolveId(element, beanDefinition, parserContext);
String beanName = resolveId(element, beanDefinition, parserContext);
parserContext.registerBeanComponent(new BeanComponentDefinition(beanDefinition, beanName));
return null;
}
@@ -187,12 +191,13 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
@SuppressWarnings("unchecked")
Collection<String> channelCandidateNames =
(Collection<String>) caValues.getArgumentValue(0, Collection.class).getValue(); // NOSONAR see comment above
(Collection<String>) caValues.getArgumentValue(0, Collection.class)
.getValue(); // NOSONAR see comment above
channelCandidateNames.add(inputChannelName); // NOSONAR
}
else {
parserContext.getReaderContext().error("Failed to locate '" +
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'",
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'",
parserContext.getRegistry());
}
}

30
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -28,6 +28,8 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.ConversionService;
@@ -136,6 +138,32 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
return null;
}
public String getBeanDescription() {
String description = null;
Object source = null;
if (this.beanFactory instanceof ConfigurableListableBeanFactory &&
((ConfigurableListableBeanFactory) this.beanFactory).containsBeanDefinition(this.beanName)) {
BeanDefinition beanDefinition =
((ConfigurableListableBeanFactory) this.beanFactory).getBeanDefinition(this.beanName);
description = beanDefinition.getResourceDescription();
source = beanDefinition.getSource();
}
StringBuilder sb = new StringBuilder("bean '")
.append(this.beanName).append("'");
if (!this.beanName.equals(getComponentName())) {
sb.append("for component '").append(getComponentName()).append("'");
}
if (description != null) {
sb.append("; defined in: '").append(description).append("'");
}
if (source != null) {
sb.append("; from source: '").append(source).append("'");
}
return sb.toString();
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
@@ -297,7 +325,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
@Override
public String toString() {
return (this.beanName != null) ? this.beanName : super.toString();
return (this.beanName != null) ? getBeanDescription() : super.toString();
}
@SuppressWarnings("unchecked")

22
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java
View File

@@ -49,6 +49,7 @@ import org.springframework.context.MessageSourceAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.io.DescriptiveResource;
import org.springframework.core.type.MethodMetadata;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
@@ -426,16 +427,25 @@ public class IntegrationFlowBeanPostProcessor
private void registerComponent(Object component, String beanName, String parentName,
BeanDefinitionCustomizer... customizers) {
BeanDefinition beanDefinition =
AbstractBeanDefinition beanDefinition =
BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) component.getClass(), () -> component)
.applyCustomizers(customizers)
.getRawBeanDefinition();
.applyCustomizers(customizers)
.getRawBeanDefinition();
if (parentName != null && this.beanFactory.containsBeanDefinition(parentName)) {
AbstractBeanDefinition parentBeanDefinition =
(AbstractBeanDefinition) this.beanFactory.getBeanDefinition(parentName);
beanDefinition.setResource(parentBeanDefinition.getResource());
Object source = parentBeanDefinition.getSource();
if (source instanceof MethodMetadata) {
source = "bean method " + ((MethodMetadata) source).getMethodName();
}
beanDefinition.setSource(source);
this.beanFactory.registerDependentBean(parentName, beanName);
}
((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition);
if (parentName != null) {
this.beanFactory.registerDependentBean(parentName, beanName);
}
this.beanFactory.getBean(beanName);
}

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java
View File

@@ -219,6 +219,16 @@ public interface IntegrationFlowContext {
*/
IntegrationFlowRegistrationBuilder addBean(String name, Object bean);
/**
* Set the configuration source {@code Object} for this manual Integration flow definition.
* Can be any arbitrary object which could easily lead to a source code for the flow when
* a messaging exception happens at runtime.
* @param source the configuration source representation.
* @return the current builder instance
* @since 5.2
*/
IntegrationFlowRegistrationBuilder setSource(Object source);
/**
* Invoke this method to prefix bean names in the flow with the (required) flow id
* and a period. This is useful if you wish to register the same flow multiple times

40
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java
View File

@@ -27,10 +27,11 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.core.type.MethodMetadata;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.support.context.NamedComponent;
@@ -105,7 +106,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
"An existing IntegrationFlowRegistration must be destroyed before overriding.");
}
integrationFlow = (IntegrationFlow) registerBean(integrationFlow, flowId, null);
integrationFlow = registerFlowBean(integrationFlow, flowId, builder.source);
}
finally {
if (registerBeanLock != null) {
@@ -124,24 +125,38 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
this.registry.put(flowId, builder.integrationFlowRegistration);
}
private IntegrationFlow registerFlowBean(IntegrationFlow flow, String flowId, @Nullable Object source) {
AbstractBeanDefinition beanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(IntegrationFlow.class, () -> flow)
.getRawBeanDefinition();
beanDefinition.setSource(source);
this.beanDefinitionRegistry.registerBeanDefinition(flowId, beanDefinition);
return this.beanFactory.getBean(flowId, IntegrationFlow.class);
}
@SuppressWarnings("unchecked")
private Object registerBean(Object bean, @Nullable String beanNameArg, String parentName) {
private void registerBean(Object bean, @Nullable String beanNameArg, String parentName) {
String beanName = beanNameArg;
if (beanName == null) {
beanName = generateBeanName(bean, parentName);
}
BeanDefinition beanDefinition =
AbstractBeanDefinition beanDefinition =
BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) bean.getClass(), () -> bean)
.getRawBeanDefinition();
this.beanDefinitionRegistry.registerBeanDefinition(beanName, beanDefinition);
if (parentName != null) {
AbstractBeanDefinition parentBeanDefinition =
(AbstractBeanDefinition) this.beanFactory.getBeanDefinition(parentName);
Object source = parentBeanDefinition.getSource();
if (source instanceof MethodMetadata) {
source = "bean method " + ((MethodMetadata) source).getMethodName();
}
beanDefinition.setSource(source);
this.beanFactory.registerDependentBean(parentName, beanName);
}
return this.beanFactory.getBean(beanName);
this.beanDefinitionRegistry.registerBeanDefinition(beanName, beanDefinition);
this.beanFactory.getBean(beanName);
}
/**
@@ -247,6 +262,9 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
private boolean idAsPrefix;
@Nullable
private Object source;
StandardIntegrationFlowRegistrationBuilder(IntegrationFlow integrationFlow) {
this.integrationFlowRegistration = new StandardIntegrationFlowRegistration(integrationFlow);
this.integrationFlowRegistration.setBeanFactory(StandardIntegrationFlowContext.this.beanFactory);
@@ -306,6 +324,12 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
return this;
}
@Override
public IntegrationFlowRegistrationBuilder setSource(Object source) {
this.source = source;
return this;
}
@Override
public IntegrationFlowRegistrationBuilder useFlowIdAsPrefix() {
this.idAsPrefix = true;

2
spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -177,7 +177,7 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
this.messagingTemplate.send(channel, message);
}
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message, "MessageFilter '" + getBeanName() + "' rejected Message");
throw new MessageRejectedException(message, "message has been rejected in filter: " + this);
}
}
return result;

67
spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java
View File

@@ -352,27 +352,34 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
}
private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) {
String errorChannel = Objects.toString(gateway.getErrorChannel(), null);
String requestChannel = Objects.toString(gateway.getRequestChannel(), null);
String errorChannel = channelToBeanName(gateway.getErrorChannel());
String requestChannel = channelToBeanName(gateway.getRequestChannel());
return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, requestChannel, errorChannel);
}
@Nullable
private String channelToBeanName(MessageChannel messageChannel) {
return messageChannel instanceof NamedComponent
? ((NamedComponent) messageChannel).getBeanName()
: Objects.toString(messageChannel, null);
}
private MessageProducerNode producerNode(String name, MessageProducerSupport producer) {
String errorChannel = Objects.toString(producer.getErrorChannel(), null);
String outputChannel = Objects.toString(producer.getOutputChannel(), null);
String errorChannel = channelToBeanName(producer.getErrorChannel());
String outputChannel = channelToBeanName(producer.getOutputChannel());
return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer,
outputChannel, errorChannel);
}
private MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) {
String errorChannel = Objects.toString(adapter.getDefaultErrorChannel(), null);
String outputChannel = Objects.toString(adapter.getOutputChannel(), null);
String errorChannel = channelToBeanName(adapter.getDefaultErrorChannel());
String outputChannel = channelToBeanName(adapter.getOutputChannel());
return new MessageSourceNode(this.nodeId.incrementAndGet(), name, adapter.getMessageSource(),
outputChannel, errorChannel);
}
private MessageHandlerNode handlerNode(String name, IntegrationConsumer consumer) {
String outputChannelName = Objects.toString(consumer.getOutputChannel(), null);
String outputChannelName = channelToBeanName(consumer.getOutputChannel());
MessageHandler handler = consumer.getHandler();
if (handler instanceof CompositeMessageHandler) {
return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null,
@@ -391,36 +398,36 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
outputChannelName, null, false);
}
else {
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String inputChannel = channelToBeanName(consumer.getInputChannel());
return new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, outputChannelName);
}
}
private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) {
String outputChannelName = Objects.toString(consumer.getOutputChannel(), null);
String errorChannel = Objects.toString(consumer.getDefaultErrorChannel(), null);
String outputChannelName = channelToBeanName(consumer.getOutputChannel());
String errorChannel = channelToBeanName(consumer.getDefaultErrorChannel());
MessageHandler handler = consumer.getHandler();
if (handler instanceof CompositeMessageHandler) {
return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName,
errorChannel, true);
errorChannel, true);
}
else if (handler instanceof DiscardingMessageHandler) {
return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName,
errorChannel, true);
errorChannel, true);
}
else if (handler instanceof MappingMessageRouterManagement) {
return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
outputChannelName, errorChannel, true);
outputChannelName, errorChannel, true);
}
else if (handler instanceof RecipientListRouterManagement) {
return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
outputChannelName, errorChannel, true);
outputChannelName, errorChannel, true);
}
else {
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String inputChannel = channelToBeanName(consumer.getInputChannel());
return new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, outputChannelName, errorChannel);
inputChannel, outputChannelName, errorChannel);
}
}
@@ -438,24 +445,24 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
named.getComponentType()))
.collect(Collectors.toList());
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String inputChannel = channelToBeanName(consumer.getInputChannel());
return polled
? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, errors, innerHandlers)
inputChannel, output, errors, innerHandlers)
: new CompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, innerHandlers);
inputChannel, output, innerHandlers);
}
private MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer,
DiscardingMessageHandler handler, String output, String errors, boolean polled) {
String discards = Objects.toString(handler.getDiscardChannel(), null);
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String discards = channelToBeanName(handler.getDiscardChannel());
String inputChannel = channelToBeanName(consumer.getInputChannel());
return polled
? new ErrorCapableDiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, discards, errors)
inputChannel, output, discards, errors)
: new DiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, discards);
inputChannel, output, discards);
}
private MessageHandlerNode routingHandler(String name, IntegrationConsumer consumer, MessageHandler handler,
@@ -466,12 +473,12 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
router.getDynamicChannelNames().stream())
.collect(Collectors.toList());
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String inputChannel = channelToBeanName(consumer.getInputChannel());
return polled
? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, errors, routes)
inputChannel, output, errors, routes)
: new RoutingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, routes);
inputChannel, output, routes);
}
private MessageHandlerNode recipientListRoutingHandler(String name, IntegrationConsumer consumer,
@@ -481,15 +488,15 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
List<String> routes =
router.getRecipients()
.stream()
.map(recipient -> ((Recipient) recipient).getChannel().toString())
.map(recipient -> channelToBeanName(((Recipient) recipient).getChannel()))
.collect(Collectors.toList());
String inputChannel = Objects.toString(consumer.getInputChannel(), null);
String inputChannel = channelToBeanName(consumer.getInputChannel());
return polled
? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, errors, routes)
inputChannel, output, errors, routes)
: new RoutingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
inputChannel, output, routes);
inputChannel, output, routes);
}
private void reset() {

12
spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
View File

@@ -30,9 +30,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.MethodIntrospector;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -100,17 +98,19 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
}
catch (InvocationTargetException e) {
if (e.getTargetException() instanceof ClassCastException) {
LOGGER.error("Could not invoke the method due to a class cast exception, " +
LOGGER.error("Could not invoke the method '" + this.method + "' due to a class cast exception, " +
"if using a lambda in the DSL, consider using an overloaded EIP method " +
"that takes a Class<?> argument to explicitly specify the type. " +
"An example of when this often occurs is if the lambda is configured to " +
"receive a Message<?> argument.", e.getCause());
}
throw new MessageHandlingException(message, e.getCause()); // NOSONAR lost stack trace
throw new IllegalStateException(
"Could not invoke the method '" + this.method + "'", e.getCause()); // NOSONAR lost stack trace
}
catch (Exception e) {
throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,
() -> "error occurred during processing message in 'LambdaMessageProcessor' [" + this + "]", e);
throw new IllegalStateException(
"error occurred during processing message in 'LambdaMessageProcessor' for method [" +
this.method + "]", e);
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -112,7 +112,8 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
if (e instanceof MessageTransformationException) {
throw (MessageTransformationException) e;
}
throw new MessageTransformationException(message, "Failed to transform Message", e);
throw new MessageTransformationException(message,
"Failed to transform Message in " + this, e);
}
}

11
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java
View File

@@ -94,7 +94,6 @@ public class AggregatorIntegrationTests {
Message<?> receive = output.receive(10000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo(1 + 2 + 3 + 4);
assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)).isEqualTo(0);
}
@Test
@@ -187,7 +186,8 @@ public class AggregatorIntegrationTests {
@Test
public void testGroupTimeoutExpressionScheduling() {
// Since group-timeout-expression="size() >= 2 ? 100 : null". The first message won't be scheduled to 'forceComplete'
// Since group-timeout-expression="size() >= 2 ? 100 : null". The first message won't be scheduled to
// 'forceComplete'
this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(1, stubHeaders(1, 6, 1)));
assertThat(this.output.receive(0)).isNull();
assertThat(this.discard.receive(0)).isNull();
@@ -239,7 +239,9 @@ public class AggregatorIntegrationTests {
ErrorMessage em = (ErrorMessage) this.errors.receive(10000);
assertThat(em).isNotNull();
assertThat(em.getPayload().getMessage().toLowerCase())
.contains("failed to send message to channel 'output' within timeout: 10");
.contains("failed to send message to channel")
.contains("output")
.contains("within timeout: 10");
}
finally {
this.output.purge(null);
@@ -261,6 +263,7 @@ public class AggregatorIntegrationTests {
}
public static class SummingAggregator {
public Integer sum(List<Integer> numbers) {
int result = 0;
for (Integer number : numbers) {
@@ -268,8 +271,8 @@ public class AggregatorIntegrationTests {
}
return result;
}
}
}

18
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java
View File

@@ -156,9 +156,9 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
assertThat(receive).isInstanceOf(ErrorMessage.class);
assertThat(receive.getPayload()).isInstanceOf(MessageRejectedException.class);
MessageRejectedException exception = (MessageRejectedException) receive.getPayload();
assertThat(exception.getMessage()).contains("MessageFilter " +
"'messagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.filter.filter.handler'" +
" rejected Message");
assertThat(exception.getMessage())
.contains("message has been rejected in filter: bean " +
"'messagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.filter.filter.handler'");
}
for (Message<?> message : this.collector) {
@@ -166,12 +166,12 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
MessageHistory messageHistory = MessageHistory.read(message);
assertThat(messageHistory).isNotNull();
String messageHistoryString = messageHistory.toString();
assertThat(messageHistoryString).contains("routerChannel");
assertThat(messageHistoryString).contains("filterChannel");
assertThat(messageHistoryString).contains("aggregatorChannel");
assertThat(messageHistoryString).contains("splitterChannel");
assertThat(messageHistoryString).contains("serviceChannel");
assertThat(messageHistoryString).doesNotContain("discardChannel");
assertThat(messageHistoryString).contains("routerChannel")
.contains("filterChannel")
.contains("aggregatorChannel")
.contains("splitterChannel")
.contains("serviceChannel")
.doesNotContain("discardChannel");
}
assertThat(this.skippedServiceActivator).isNull();

85
spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherParserTests.java
View File

@@ -17,9 +17,9 @@
package org.springframework.integration.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
@@ -30,68 +30,77 @@ import org.springframework.integration.transformer.MessageTransformationExceptio
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class HeaderEnricherParserTests {
@SpringJUnitConfig
class HeaderEnricherParserTests {
@Autowired
private ApplicationContext context;
@Test // INT-1154
public void sendTimeoutDefault() {
@Test
void sendTimeoutDefault() {
Object endpoint = context.getBean("headerEnricherWithDefaults");
long sendTimeout = TestUtils.getPropertyValue(endpoint, "handler.messagingTemplate.sendTimeout", Long.class);
assertThat(sendTimeout).isEqualTo(-1L);
}
@Test // INT-1154
public void sendTimeoutConfigured() {
@Test
void sendTimeoutConfigured() {
Object endpoint = context.getBean("headerEnricherWithSendTimeout");
long sendTimeout = TestUtils.getPropertyValue(endpoint, "handler.messagingTemplate.sendTimeout", Long.class);
assertThat(sendTimeout).isEqualTo(1234L);
}
@Test // INT-1167
public void shouldSkipNullsDefault() {
@Test
void shouldSkipNullsDefault() {
Object endpoint = context.getBean("headerEnricherWithDefaults");
Boolean shouldSkipNulls = TestUtils.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
Boolean shouldSkipNulls = TestUtils
.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
assertThat(shouldSkipNulls).isEqualTo(Boolean.TRUE);
}
@Test // INT-1167
public void shouldSkipNullsFalseConfigured() {
Object endpoint = context.getBean("headerEnricherWithShouldSkipNullsFalse");
Boolean shouldSkipNulls = TestUtils.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
assertThat(shouldSkipNulls).isEqualTo(Boolean.FALSE);
}
@Test // INT-1167
public void shouldSkipNullsTrueConfigured() {
Object endpoint = context.getBean("headerEnricherWithShouldSkipNullsTrue");
Boolean shouldSkipNulls = TestUtils.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
assertThat(shouldSkipNulls).isEqualTo(Boolean.TRUE);
}
@Test(expected = MessageTransformationException.class)
public void testStringPriorityHeader() {
MessageHandler messageHandler =
TestUtils.getPropertyValue(context.getBean("headerEnricherWithPriorityAsString"), "handler", MessageHandler.class);
Message<?> message = new GenericMessage<String>("hello");
messageHandler.handleMessage(message);
}
@Test
public void testStringPriorityHeaderWithType() {
void shouldSkipNullsFalseConfigured() {
Object endpoint = context.getBean("headerEnricherWithShouldSkipNullsFalse");
Boolean shouldSkipNulls = TestUtils
.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
assertThat(shouldSkipNulls).isEqualTo(Boolean.FALSE);
}
@Test
void shouldSkipNullsTrueConfigured() {
Object endpoint = context.getBean("headerEnricherWithShouldSkipNullsTrue");
Boolean shouldSkipNulls = TestUtils
.getPropertyValue(endpoint, "handler.transformer.shouldSkipNulls", Boolean.class);
assertThat(shouldSkipNulls).isEqualTo(Boolean.TRUE);
}
@Test
void testStringPriorityHeader() {
MessageHandler messageHandler =
TestUtils.getPropertyValue(context.getBean("headerEnricherWithPriorityAsStringAndType"), "handler", MessageHandler.class);
TestUtils.getPropertyValue(this.context.getBean("headerEnricherWithPriorityAsString"),
"handler", MessageHandler.class);
Message<?> message = new GenericMessage<>("hello");
assertThatExceptionOfType(MessageTransformationException.class)
.isThrownBy(() -> messageHandler.handleMessage(message))
.withMessageContaining(
"; defined in: 'class path resource " +
"[org/springframework/integration/config/xml/HeaderEnricherParserTests-context.xml]'");
}
@Test
void testStringPriorityHeaderWithType() {
MessageHandler messageHandler =
TestUtils.getPropertyValue(context.getBean("headerEnricherWithPriorityAsStringAndType"),
"handler", MessageHandler.class);
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build();
messageHandler.handleMessage(message);

13
spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java
View File

@@ -31,7 +31,6 @@ import org.springframework.integration.handler.LambdaMessageProcessor;
import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
@@ -61,12 +60,12 @@ public class LambdaMessageProcessorTests {
LambdaMessageProcessor lmp = new LambdaMessageProcessor(
new GenericTransformer<Message<?>, Message<?>>() { // Must not be lambda
@Override
public Message<?> transform(Message<?> source) {
return messageTransformer(source);
}
@Override
public Message<?> transform(Message<?> source) {
return messageTransformer(source);
}
}, null);
}, null);
lmp.setBeanFactory(mock(BeanFactory.class));
GenericMessage<String> testMessage = new GenericMessage<>("foo");
Object result = lmp.processMessage(testMessage);
@@ -79,7 +78,7 @@ public class LambdaMessageProcessorTests {
(GenericTransformer<Message<?>, Message<?>>) this::messageTransformer, null);
lmp.setBeanFactory(mock(BeanFactory.class));
GenericMessage<String> testMessage = new GenericMessage<>("foo");
assertThatExceptionOfType(MessageHandlingException.class)
assertThatExceptionOfType(IllegalStateException.class)
.isThrownBy(() -> lmp.processMessage(testMessage))
.withCauseInstanceOf(ClassCastException.class);
}

35
spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.dsl.flows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.fail;
import java.io.Serializable;
@@ -40,7 +41,6 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
@@ -233,15 +233,11 @@ public class IntegrationFlowTests {
assertThat(this.beanFactory.containsBean("bridgeFlow2.channel#0")).isTrue();
assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0")).isInstanceOf(FixedSubscriberChannel.class);
try {
this.bridgeFlow2Input.send(message);
fail("Expected MessageDispatchingException");
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(MessageDispatchingException.class);
assertThat(e.getMessage()).contains("Dispatcher has no subscribers");
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> this.bridgeFlow2Input.send(message))
.withCauseInstanceOf(MessageDispatchingException.class)
.withMessageContaining("Dispatcher has no subscribers");
this.controlBus.send("@bridge.start()");
this.bridgeFlow2Input.send(message);
reply = this.bridgeFlow2Output.receive(10000);
@@ -252,21 +248,10 @@ public class IntegrationFlowTests {
@Test
public void testWrongLastMessageChannel() {
ConfigurableApplicationContext context = null;
try {
context = new AnnotationConfigApplicationContext(InvalidLastMessageChannelFlowContext.class);
fail("BeanCreationException expected");
}
catch (Exception e) {
assertThat(e).isInstanceOf(BeanCreationException.class);
assertThat(e.getMessage()).contains("'.fixedSubscriberChannel()' " +
"can't be the last EIP-method in the 'IntegrationFlow' definition");
}
finally {
if (context != null) {
context.close();
}
}
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> new AnnotationConfigApplicationContext(InvalidLastMessageChannelFlowContext.class))
.withMessageContaining("'.fixedSubscriberChannel()' " +
"can't be the last EIP-method in the 'IntegrationFlow' definition");
}
@Test

9
spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java
View File

@@ -78,7 +78,12 @@ public class GatewayDslTests {
assertThat(receive).isNotNull();
assertThat(receive).isInstanceOf(ErrorMessage.class);
assertThat(receive.getPayload()).isInstanceOf(MessageRejectedException.class);
assertThat(((Exception) receive.getPayload()).getMessage()).contains("' rejected Message");
String exceptionMessage = ((Exception) receive.getPayload()).getMessage();
assertThat(exceptionMessage)
.contains("message has been rejected in filter")
.contains("defined in: " +
"'org.springframework.integration.dsl.gateway.GatewayDslTests$ContextConfiguration'; " +
"from source: 'bean method gatewayRequestFlow'");
}
@Autowired
@@ -89,7 +94,7 @@ public class GatewayDslTests {
void testNestedGatewayErrorPropagation() {
assertThatExceptionOfType(RuntimeException.class)
.isThrownBy(() -> this.nestedGatewayErrorPropagationFlowInput.send(new GenericMessage<>("test")))
.withMessageContaining("intentional");
.withStackTraceContaining("intentional");
}
@Configuration

21
spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java
View File

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.fail;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
@@ -73,6 +74,7 @@ import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.support.SmartLifecycleRoleController;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.util.NoBeansOverrideAnnotationConfigContextLoader;
import org.springframework.messaging.Message;
@@ -86,6 +88,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ReflectionUtils;
import reactor.core.publisher.Flux;
@@ -521,6 +524,24 @@ public class ManualFlowTests {
.withMessageContaining("Invalid bean definition with name 'doNotOverrideChannel'");
}
@Test
public void testBeanDefinitionInfoInTheException() {
IntegrationFlow testFlow = f -> f.<String, String>transform(String::toUpperCase);
Method source = ReflectionUtils.findMethod(ManualFlowTests.class, "testBeanDefinitionInfoInTheException");
IntegrationFlowRegistration flowRegistration =
this.integrationFlowContext.registration(testFlow)
.setSource(source)
.register();
assertThatExceptionOfType(MessageTransformationException.class)
.isThrownBy(() -> flowRegistration.getInputChannel().send(new GenericMessage<>(new Date())))
.withCauseExactlyInstanceOf(IllegalStateException.class)
.withRootCauseInstanceOf(ClassCastException.class)
.withMessageContaining("from source: '" + source + "'")
.withStackTraceContaining("java.util.Date cannot be cast to java.lang.String");
flowRegistration.destroy();
}
@Configuration
@EnableIntegration
@EnableMessageHistory

22
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java
View File

@@ -123,7 +123,7 @@ public class AdvisedMessageHandlerTests {
handler.setComponentName(componentName);
QueueChannel replies = new QueueChannel();
handler.setOutputChannel(replies);
Message<String> message = new GenericMessage<String>("Hello, world!");
Message<String> message = new GenericMessage<>("Hello, world!");
// no advice
handler.handleMessage(message);
@@ -142,7 +142,7 @@ public class AdvisedMessageHandlerTests {
List<Advice> adviceChain = new ArrayList<Advice>();
adviceChain.add(advice);
final AtomicReference<String> compName = new AtomicReference<String>();
final AtomicReference<String> compName = new AtomicReference<>();
adviceChain.add(new AbstractRequestHandlerAdvice() {
@Override
@@ -229,7 +229,7 @@ public class AdvisedMessageHandlerTests {
};
QueueChannel replies = new QueueChannel();
handler.setOutputChannel(replies);
Message<String> message = new GenericMessage<String>("Hello, world!");
Message<String> message = new GenericMessage<>("Hello, world!");
PollableChannel successChannel = new QueueChannel();
PollableChannel failureChannel = new QueueChannel();
@@ -293,7 +293,7 @@ public class AdvisedMessageHandlerTests {
};
QueueChannel replies = new QueueChannel();
handler.setOutputChannel(replies);
Message<String> message = new GenericMessage<String>("Hello, world!");
Message<String> message = new GenericMessage<>("Hello, world!");
PollableChannel successChannel = new QueueChannel();
PollableChannel failureChannel = new QueueChannel();
@@ -375,7 +375,7 @@ public class AdvisedMessageHandlerTests {
advice.setThreshold(2);
advice.setHalfOpenAfter(1000);
List<Advice> adviceChain = new ArrayList<Advice>();
List<Advice> adviceChain = new ArrayList<>();
adviceChain.add(advice);
handler.setAdviceChain(adviceChain);
handler.setBeanFactory(mock(BeanFactory.class));
@@ -402,7 +402,7 @@ public class AdvisedMessageHandlerTests {
fail("Expected failure");
}
catch (Exception e) {
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for baz");
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for bean 'baz'");
assertThat(((MessagingException) e).getFailedMessage()).isSameAs(message);
}
@@ -426,7 +426,7 @@ public class AdvisedMessageHandlerTests {
fail("Expected failure");
}
catch (Exception e) {
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for baz");
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for bean 'baz'");
}
// Simulate some timeout in between requests
@@ -454,7 +454,7 @@ public class AdvisedMessageHandlerTests {
fail("Expected failure");
}
catch (Exception e) {
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for baz");
assertThat(e.getMessage()).isEqualTo("Circuit Breaker is Open for bean 'baz'");
}
}
@@ -475,7 +475,7 @@ public class AdvisedMessageHandlerTests {
handler.setOutputChannel(replies);
RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
List<Advice> adviceChain = new ArrayList<Advice>();
List<Advice> adviceChain = new ArrayList<>();
adviceChain.add(advice);
handler.setAdviceChain(adviceChain);
handler.setBeanFactory(mock(BeanFactory.class));
@@ -937,7 +937,7 @@ public class AdvisedMessageHandlerTests {
MessageFilter filter = new MessageFilter(message -> false);
final QueueChannel discardChannel = new QueueChannel();
filter.setDiscardChannel(discardChannel);
List<Advice> adviceChain = new ArrayList<Advice>();
List<Advice> adviceChain = new ArrayList<>();
final AtomicReference<Message<?>> discardedWithinAdvice = new AtomicReference<Message<?>>();
adviceChain.add(new AbstractRequestHandlerAdvice() {
@@ -1017,7 +1017,7 @@ public class AdvisedMessageHandlerTests {
RetryTemplate retryTemplate = new RetryTemplate();
Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<Class<? extends Throwable>, Boolean>();
Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
retryableExceptions.put(MyException.class, retryForMyException);
retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3, retryableExceptions, true));

27
spring-integration-core/src/test/java/org/springframework/integration/transformer/ContentEnricherTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.transformer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.Mockito.mock;
@@ -143,7 +144,8 @@ public class ContentEnricherTests {
}
catch (ReplyRequiredException e) {
assertThat(e.getMessage())
.isEqualTo("No reply produced by handler 'Enricher', and its 'requiresReply' property is set to true.");
.isEqualTo("No reply produced by handler 'Enricher', and its 'requiresReply' property is set to " +
"true.");
return;
}
finally {
@@ -174,14 +176,11 @@ public class ContentEnricherTests {
Target target = new Target("replace me");
Message<?> requestMessage = MessageBuilder.withPayload(target).setReplyChannel(replyChannel).build();
try {
enricher.handleMessage(requestMessage);
}
catch (MessageDeliveryException e) {
assertThat(e.getMessage()).isEqualToIgnoringCase("failed to send message to channel '" + requestChannelName
+ "' within timeout: " + requestTimeout);
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> enricher.handleMessage(requestMessage))
.withMessageContaining("Failed to send message to channel")
.withMessageContaining(requestChannelName)
.withMessageContaining("within timeout: " + requestTimeout);
}
@Test
@@ -189,6 +188,7 @@ public class ContentEnricherTests {
QueueChannel replyChannel = new QueueChannel();
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -199,7 +199,7 @@ public class ContentEnricherTests {
enricher.setRequestChannel(requestChannel);
SpelExpressionParser parser = new SpelExpressionParser();
Map<String, Expression> propertyExpressions = new HashMap<String, Expression>();
Map<String, Expression> propertyExpressions = new HashMap<>();
propertyExpressions.put("name", parser.parseExpression("payload.lastName + ', ' + payload.firstName"));
enricher.setPropertyExpressions(propertyExpressions);
enricher.setBeanFactory(mock(BeanFactory.class));
@@ -308,6 +308,7 @@ public class ContentEnricherTests {
QueueChannel replyChannel = new QueueChannel();
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -337,6 +338,7 @@ public class ContentEnricherTests {
QueueChannel replyChannel = new QueueChannel();
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -367,6 +369,7 @@ public class ContentEnricherTests {
QueueChannel replyChannel = new QueueChannel();
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -400,6 +403,7 @@ public class ContentEnricherTests {
QueueChannel replyChannel = new QueueChannel();
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -450,6 +454,7 @@ public class ContentEnricherTests {
DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Source("John", "Doe");
@@ -483,6 +488,7 @@ public class ContentEnricherTests {
final DirectChannel requestChannel = new DirectChannel();
requestChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
throw new RuntimeException();
@@ -492,6 +498,7 @@ public class ContentEnricherTests {
final DirectChannel errorChannel = new DirectChannel();
errorChannel.subscribe(new AbstractReplyProducingMessageHandler() {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return new Target("failed");

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java
View File

@@ -289,7 +289,7 @@ public class ConnectionEventTests {
String actual = theEvent.toString();
assertThat(actual).contains("cause=java.net.BindException");
assertThat(actual).contains("source="
+ "sf, port=" + factory.getPort());
+ "bean 'sf', port=" + factory.getPort());
ArgumentCaptor<String> reasonCaptor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<Throwable> throwableCaptor = ArgumentCaptor.forClass(Throwable.class);

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java
View File

@@ -240,7 +240,7 @@ public class ConnectionFactoryTests {
assertThat(n < 200).as("Stop was not invoked in time").isTrue();
latch2.countDown();
assertThat(latch3.await(10, TimeUnit.SECONDS)).as("missing debug log").isTrue();
String expected = "foo, port=" + factory.getPort() + message;
String expected = "bean 'foo', port=" + factory.getPort() + message;
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
verify(logger, atLeast(1)).debug(captor.capture());
assertThat(captor.getAllValues()).contains(expected);

8
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -1017,7 +1017,13 @@ public class IntegrationMBeanExporter extends MBeanExporter
* Short hack to makes sure object names are unique if more than one endpoint has the same input
* channel
*/
return messageChannel + (total > 1 ? "#" + total : "");
String channelName =
messageChannel instanceof NamedComponent
? ((NamedComponent) messageChannel).getBeanName()
: messageChannel.toString();
return channelName + (total > 1 ? "#" + total : "");
}
/**

26
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java
View File

@@ -22,6 +22,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.BaseHandlerMetrics;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
@@ -35,6 +36,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Dave Syer
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
@ContextConfiguration
@@ -54,28 +56,27 @@ public class ChannelIntegrationTests {
@Autowired
private IntegrationMBeanExporter messageChannelsMonitor;
@SuppressWarnings("deprecation")
@Test
public void testMessageChannelStatistics() throws Exception {
public void testMessageChannelStatistics() {
this.requests.send(new GenericMessage<>("foo"));
requests.send(new GenericMessage<String>("foo"));
String intermediateChannelName = "" + intermediate;
String intermediateChannelName = ((NamedComponent) this.intermediate).getBeanName();
assertThat(messageChannelsMonitor.getChannelSendCount(intermediateChannelName)).isEqualTo(1);
double rate = messageChannelsMonitor.getChannelSendRate("" + requests).getMean();
assertThat(rate >= 0).as("No statistics for requests channel").isTrue();
double rate =
messageChannelsMonitor.getChannelSendRate(((NamedComponent) this.requests).getBeanName()).getMean();
assertThat(rate).as("No statistics for requests channel").isGreaterThanOrEqualTo(0);
rate = messageChannelsMonitor.getChannelSendRate(intermediateChannelName).getMean();
assertThat(rate >= 0).as("No statistics for intermediate channel").isTrue();
assertThat(rate).as("No statistics for intermediate channel").isGreaterThanOrEqualTo(0);
assertThat(intermediate.receive(100L)).isNotNull();
assertThat(messageChannelsMonitor.getChannelReceiveCount(intermediateChannelName)).isEqualTo(1);
requests.send(new GenericMessage<String>("foo"));
requests.send(new GenericMessage<>("foo"));
try {
requests.send(new GenericMessage<String>("foo"));
requests.send(new GenericMessage<>("foo"));
}
catch (@SuppressWarnings("unused") MessageDeliveryException e) {
}
@@ -94,9 +95,8 @@ public class ChannelIntegrationTests {
assertThat(this.sourceChannel.receive(10000)).isNotNull();
assertThat(messageChannelsMonitor.getSourceMessageCount("source") > 0).isTrue();
assertThat(messageChannelsMonitor.getSourceMetrics("source").getMessageCount() > 0).isTrue();
assertThat(messageChannelsMonitor.getSourceMessageCount("source")).isGreaterThan(0);
assertThat(messageChannelsMonitor.getSourceMetrics("source").getMessageCount()).isGreaterThan(0);
}
}

2
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java
View File

@@ -38,7 +38,7 @@ import org.springframework.messaging.support.GenericMessage;
*/
public class HandlerMonitoringIntegrationTests {
private static Log logger = LogFactory.getLog(HandlerMonitoringIntegrationTests.class);
private static final Log logger = LogFactory.getLog(HandlerMonitoringIntegrationTests.class);
private MessageChannel channel;

28
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java
View File

@@ -30,7 +30,9 @@ import org.junit.Test;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
@@ -48,7 +50,7 @@ public class MessageChannelsMonitorIntegrationTests {
private static Log logger = LogFactory.getLog(MessageChannelsMonitorIntegrationTests.class);
private MessageChannel channel;
private AbstractMessageChannel channel;
private Service service;
@@ -75,7 +77,7 @@ public class MessageChannelsMonitorIntegrationTests {
@Test
public void testRates() throws Exception {
try (ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml")) {
this.channel = context.getBean("anonymous", MessageChannel.class);
this.channel = context.getBean("anonymous", AbstractMessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(50);
service.setLatch(latch);
@@ -87,9 +89,9 @@ public class MessageChannelsMonitorIntegrationTests {
assertThat(service.getCounter()).isEqualTo(before + 50);
// The handler monitor is registered under the endpoint id (since it is explicit)
int sends = messageChannelsMonitor.getChannelSendRate("" + channel).getCount();
int sends = messageChannelsMonitor.getChannelSendRate(this.channel.getBeanName()).getCount();
assertThat(sends).as("No send statistics for input channel").isEqualTo(50);
long sendsLong = messageChannelsMonitor.getChannelSendRate("" + channel).getCountLong();
long sendsLong = messageChannelsMonitor.getChannelSendRate(this.channel.getBeanName()).getCountLong();
assertThat(sends).as("No send statistics for input channel").isEqualTo(sendsLong);
}
@@ -98,7 +100,7 @@ public class MessageChannelsMonitorIntegrationTests {
@Test
public void testErrors() throws Exception {
try (ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml")) {
this.channel = context.getBean("anonymous", MessageChannel.class);
this.channel = context.getBean("anonymous", AbstractMessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(10);
service.setLatch(latch);
@@ -120,9 +122,9 @@ public class MessageChannelsMonitorIntegrationTests {
assertThat(service.getCounter()).isEqualTo(before + 10);
// The handler monitor is registered under the endpoint id (since it is explicit)
int sends = messageChannelsMonitor.getChannelSendRate("" + channel).getCount();
int sends = messageChannelsMonitor.getChannelSendRate(this.channel.getBeanName()).getCount();
assertThat(sends).as("No send statistics for input channel").isEqualTo(11);
int errors = messageChannelsMonitor.getChannelErrorRate("" + channel).getCount();
int errors = messageChannelsMonitor.getChannelErrorRate(this.channel.getBeanName()).getCount();
assertThat(errors).as("No error statistics for input channel").isEqualTo(1);
}
}
@@ -130,7 +132,7 @@ public class MessageChannelsMonitorIntegrationTests {
@Test
public void testQueues() throws Exception {
try (ClassPathXmlApplicationContext context = createContext("queue-channel.xml")) {
this.channel = context.getBean("queue", MessageChannel.class);
this.channel = context.getBean("queue", AbstractMessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(10);
service.setLatch(latch);
@@ -152,18 +154,18 @@ public class MessageChannelsMonitorIntegrationTests {
assertThat(service.getCounter()).isEqualTo(before + 10);
// The handler monitor is registered under the endpoint id (since it is explicit)
int sends = messageChannelsMonitor.getChannelSendRate("" + channel).getCount();
int sends = messageChannelsMonitor.getChannelSendRate(this.channel.getBeanName()).getCount();
assertThat(sends).as("No send statistics for input channel").isEqualTo(11);
int receives = messageChannelsMonitor.getChannelReceiveCount("" + channel);
int receives = messageChannelsMonitor.getChannelReceiveCount(this.channel.getBeanName());
assertThat(receives).as("No send statistics for input channel").isEqualTo(11);
int errors = messageChannelsMonitor.getChannelErrorRate("" + channel).getCount();
int errors = messageChannelsMonitor.getChannelErrorRate(this.channel.getBeanName()).getCount();
assertThat(errors).as("Expect no errors for input channel (handler fails)").isEqualTo(0);
}
}
private void doTest(String config, String channelName) throws Exception {
try (ClassPathXmlApplicationContext context = createContext(config)) {
this.channel = context.getBean(channelName, MessageChannel.class);
MessageChannel channel = context.getBean(channelName, MessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(1);
service.setLatch(latch);
@@ -172,7 +174,7 @@ public class MessageChannelsMonitorIntegrationTests {
assertThat(service.getCounter()).isEqualTo(before + 1);
// The handler monitor is registered under the endpoint id (since it is explicit)
int sends = messageChannelsMonitor.getChannelSendRate("" + channel).getCount();
int sends = messageChannelsMonitor.getChannelSendRate(((NamedComponent) channel).getBeanName()).getCount();
assertThat(sends).as("No statistics for input channel").isEqualTo(1);
assertThat(channel).isInstanceOf(InterceptableChannel.class);