diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java index 5a722ae268..8a00576ce4 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java @@ -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 allContainerAttributes = new ArrayList(Arrays.asList(CONTAINER_VALUE_ATTRIBUTES)); + List 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); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java index a4ba3d0db7..f9642e6f22 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java @@ -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 ) 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 channelCandidateNames = - (Collection) caValues.getArgumentValue(0, Collection.class).getValue(); // NOSONAR see comment above + (Collection) 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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 1557b240de..76f503ce97 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -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") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java index 7673279d8c..6796194fc7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java @@ -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) 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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java index 121278a48f..03f42be2f7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java @@ -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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java index a8d42b5beb..3ba73336a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java @@ -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) 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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index 6a4db0b02a..730bbdae14 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java index 8d6e792cd1..fbc0ba705b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java @@ -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 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() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java index e607feb923..56ca5d8c4a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java @@ -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, 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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index f6c9cb9434..6b0803a0de 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -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); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index 8ba0891c87..b3f8429c5b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -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 numbers) { int result = 0; for (Integer number : numbers) { @@ -268,8 +271,8 @@ public class AggregatorIntegrationTests { } return result; } + } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 64005297b0..0ee68dfc04 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherParserTests.java index b69220dc84..da1f252567 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherParserTests.java @@ -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("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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java index 14d7d7c28f..eecd013265 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java @@ -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>() { // 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 testMessage = new GenericMessage<>("foo"); Object result = lmp.processMessage(testMessage); @@ -79,7 +78,7 @@ public class LambdaMessageProcessorTests { (GenericTransformer, Message>) this::messageTransformer, null); lmp.setBeanFactory(mock(BeanFactory.class)); GenericMessage testMessage = new GenericMessage<>("foo"); - assertThatExceptionOfType(MessageHandlingException.class) + assertThatExceptionOfType(IllegalStateException.class) .isThrownBy(() -> lmp.processMessage(testMessage)) .withCauseInstanceOf(ClassCastException.class); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 2fdcd7aa78..22dfe87f83 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java index bc5596372d..d027572f92 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index 4b38100ae6..15a5d6d3a4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java @@ -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.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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 22f7683ba3..4e83ca53fd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java @@ -123,7 +123,7 @@ public class AdvisedMessageHandlerTests { handler.setComponentName(componentName); QueueChannel replies = new QueueChannel(); handler.setOutputChannel(replies); - Message message = new GenericMessage("Hello, world!"); + Message message = new GenericMessage<>("Hello, world!"); // no advice handler.handleMessage(message); @@ -142,7 +142,7 @@ public class AdvisedMessageHandlerTests { List adviceChain = new ArrayList(); adviceChain.add(advice); - final AtomicReference compName = new AtomicReference(); + final AtomicReference compName = new AtomicReference<>(); adviceChain.add(new AbstractRequestHandlerAdvice() { @Override @@ -229,7 +229,7 @@ public class AdvisedMessageHandlerTests { }; QueueChannel replies = new QueueChannel(); handler.setOutputChannel(replies); - Message message = new GenericMessage("Hello, world!"); + Message 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 message = new GenericMessage("Hello, world!"); + Message 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 adviceChain = new ArrayList(); + List 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 adviceChain = new ArrayList(); + List 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 adviceChain = new ArrayList(); + List adviceChain = new ArrayList<>(); final AtomicReference> discardedWithinAdvice = new AtomicReference>(); adviceChain.add(new AbstractRequestHandlerAdvice() { @@ -1017,7 +1017,7 @@ public class AdvisedMessageHandlerTests { RetryTemplate retryTemplate = new RetryTemplate(); - Map, Boolean> retryableExceptions = new HashMap, Boolean>(); + Map, Boolean> retryableExceptions = new HashMap<>(); retryableExceptions.put(MyException.class, retryForMyException); retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3, retryableExceptions, true)); 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 4fd004e2ef..cec4c66d04 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 @@ -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 propertyExpressions = new HashMap(); + Map 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"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 8072891e20..97ef8cce0d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java @@ -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 reasonCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java index c6b537edbd..f1634bcf3f 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java @@ -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 captor = ArgumentCaptor.forClass(String.class); verify(logger, atLeast(1)).debug(captor.capture()); assertThat(captor.getAllValues()).contains(expected); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index ccdbd82cce..cefd25fe7f 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -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 : ""); } /** diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java index 8448d58443..2170d9600d 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java @@ -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("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("foo")); + requests.send(new GenericMessage<>("foo")); try { - requests.send(new GenericMessage("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); } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java index e12875a187..1b697aca92 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java @@ -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; diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java index b4172016bd..eeec1db0f8 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java @@ -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);