From a30809be3843d83faaaae92668617157b2bb5a19 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 26 May 2016 14:22:50 -0400 Subject: [PATCH] INT-4042: Support Known Router Channels in Graph JIRA: https://jira.spring.io/browse/INT-4042 ``` ... }, { "nodeId" : 24, "name" : "integrationGraphServerTests.Config.router.router", "stats" : { ... } }, "componentType" : "router", "output" : "discards", "input" : "four", "routes" : [ "barChannel", "bazChannel" ], "errors" : "myErrors" }, { ... "from" : 3, "to" : 24, "type" : "input" }, { "from" : 24, "to" : 7, "type" : "output" }, { "from" : 24, "to" : 2, "type" : "error" }, { "from" : 24, "to" : 5, "type" : "route" }, { "from" : 24, "to" : 6, "type" : "route" ... ``` Add Graph Support for RecipientListRouter The RLR is not an AMMR; it's the only router outside of that class hierarchy. Polishing Add Dynamically Routed-to Channels to Graph Add ExpressionBased and Expose Expression in Graph Polishing Add getExpressionString() to IOS Add ExpressionCapable; Set Primary Expression Polishing - PR Comments Router/Expression Docs * Simple Java Docs polishing --- .../config/ServiceActivatorFactoryBean.java | 4 +- .../config/TransformerFactoryBean.java | 4 +- .../context/ExpressionCapable.java | 37 ++++++ .../context/IntegrationObjectSupport.java | 19 ++- .../endpoint/EventDrivenConsumer.java | 4 + .../ExpressionEvaluatingMessageSource.java | 8 +- .../integration/endpoint/PollingConsumer.java | 4 + .../endpoint/SourcePollingChannelAdapter.java | 4 + .../ExpressionEvaluatingMessageHandler.java | 9 +- .../router/AbstractMappingMessageRouter.java | 41 +++++++ .../router/AbstractMessageRouter.java | 29 +++-- .../router/ExpressionEvaluatingRouter.java | 4 +- .../integration/router/MessageRouter.java | 36 ++++++ .../ExpressionEvaluatingSplitter.java | 3 +- .../MappingMessageRouterManagement.java | 16 ++- .../support/management/RouterMetrics.java | 6 + .../graph/ErrorCapableRoutingNode.java | 46 +++++++ .../graph/IntegrationGraphServer.java | 113 ++++++++++++++---- .../management/graph/IntegrationNode.java | 17 +++ .../support/management/graph/LinkNode.java | 2 +- .../graph/RoutingMessageHandlerNode.java | 44 +++++++ .../router/HeaderValueRouterTests.java | 40 ++++++- .../graph/IntegrationGraphServerTests.java | 100 +++++++++++++++- .../AbstractRemoteFileOutboundGateway.java | 4 +- .../integration/jms/JmsOutboundGateway.java | 1 + .../jms/JmsSendingMessageHandler.java | 1 + src/reference/asciidoc/graph.adoc | 6 + 27 files changed, 552 insertions(+), 50 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/ExpressionCapable.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/router/MessageRouter.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableRoutingNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/RoutingMessageHandlerNode.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java index 0c3d67948a..4def1768ad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java @@ -98,7 +98,9 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF protected MessageHandler createExpressionEvaluatingHandler(Expression expression) { ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor(expression); processor.setBeanFactory(this.getBeanFactory()); - return this.configureHandler(new ServiceActivatingHandler(processor)); + ServiceActivatingHandler handler = new ServiceActivatingHandler(processor); + handler.setPrimaryExpression(expression); + return this.configureHandler(handler); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java index f750cc915a..075bd1a98a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java @@ -63,7 +63,9 @@ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactor @Override protected MessageHandler createExpressionEvaluatingHandler(Expression expression) { Transformer transformer = new ExpressionEvaluatingTransformer(expression); - return this.createHandler(transformer); + MessageTransformingHandler handler = this.createHandler(transformer); + handler.setPrimaryExpression(expression); + return handler; } protected MessageTransformingHandler createHandler(Transformer transformer) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/ExpressionCapable.java b/spring-integration-core/src/main/java/org/springframework/integration/context/ExpressionCapable.java new file mode 100644 index 0000000000..16889ddb68 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/ExpressionCapable.java @@ -0,0 +1,37 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.context; + +import org.springframework.expression.Expression; + +/** + * Components that implement this interface are capable of supporting a primary + * SpEL expression as part of their configuration. + * + * @author Gary Russell + * @since 4.3 + * + */ +public interface ExpressionCapable { + + /** + * Return the primary SpEL expression if this component is expression-based. + * @return the expression as a String. + */ + Expression getExpression(); + +} 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 da857aee46..7fc4d4eb05 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 @@ -33,6 +33,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; +import org.springframework.expression.Expression; import org.springframework.expression.ExpressionParser; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.support.DefaultMessageBuilderFactory; @@ -63,7 +64,7 @@ import org.springframework.util.StringUtils; * @author Artem Bilan */ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedComponent, - ApplicationContextAware, BeanFactoryAware, InitializingBean { + ApplicationContextAware, BeanFactoryAware, InitializingBean, ExpressionCapable { protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); @@ -92,6 +93,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo private volatile MessageBuilderFactory messageBuilderFactory; + private Expression expression; + @Override public final void setBeanName(String beanName) { this.beanName = beanName; @@ -144,6 +147,20 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo this.channelResolver = channelResolver; } + @Override + public Expression getExpression() { + return this.expression; + } + + /** + * For expression-based components, set the primary expression. + * @param expression the expression. + * @since 4.3 + */ + public final void setPrimaryExpression(Expression expression) { + this.expression = expression; + } + @Override public final void afterPropertiesSet() { this.integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index 3edc2fdf93..fab4580f1d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -19,6 +19,7 @@ package org.springframework.integration.endpoint; import org.springframework.context.Lifecycle; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.router.MessageRouter; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -58,6 +59,9 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration if (this.handler instanceof MessageProducer) { return ((MessageProducer) this.handler).getOutputChannel(); } + else if (this.handler instanceof MessageRouter) { + return ((MessageRouter) this.handler).getDefaultOutputChannel(); + } else { return null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java index 47e26b032e..1f9c68b62e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionEvaluatingMessageSource.java @@ -17,6 +17,7 @@ package org.springframework.integration.endpoint; import org.springframework.expression.Expression; +import org.springframework.integration.context.ExpressionCapable; import org.springframework.util.Assert; /** @@ -24,7 +25,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.0 */ -public class ExpressionEvaluatingMessageSource extends AbstractMessageSource { +public class ExpressionEvaluatingMessageSource extends AbstractMessageSource implements ExpressionCapable { private final Expression expression; @@ -47,4 +48,9 @@ public class ExpressionEvaluatingMessageSource extends AbstractMessageSource< return this.evaluateExpression(this.expression, this.expectedType); } + @Override + public Expression getExpression() { + return this.expression; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 1b4c2bfb29..1f7ce78595 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -24,6 +24,7 @@ import java.util.List; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ExecutorChannelInterceptorAware; import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.router.MessageRouter; import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -83,6 +84,9 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat if (this.handler instanceof MessageProducer) { return ((MessageProducer) this.handler).getOutputChannel(); } + else if (this.handler instanceof MessageRouter) { + return ((MessageRouter) this.handler).getDefaultOutputChannel(); + } else { return null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 25012ed580..8df0679779 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -26,6 +26,7 @@ import org.springframework.aop.support.AopUtils; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.context.Lifecycle; import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.context.ExpressionCapable; import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; @@ -66,6 +67,9 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint */ public void setSource(MessageSource source) { this.source = source; + if (source instanceof ExpressionCapable) { + setPrimaryExpression(((ExpressionCapable) source).getExpression()); + } } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageHandler.java index 33ec04f2fc..889ef630c2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,6 +25,7 @@ import org.springframework.util.Assert; * the provided {@link Expression} expecting a void return. * * @author Artem Bilan + * @author Gary Russell * @see MethodInvokingMessageHandler * @since 2.1 */ @@ -36,9 +37,9 @@ public class ExpressionEvaluatingMessageHandler extends AbstractMessageHandler { public ExpressionEvaluatingMessageHandler(Expression expression) { - Assert.notNull(expression, "Expression must not be null"); - this.processor = new ExpressionEvaluatingMessageProcessor( - expression, Void.class); + Assert.notNull(expression, "'expression' must not be null"); + this.processor = new ExpressionEvaluatingMessageProcessor(expression, Void.class); + setPrimaryExpression(expression); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java index bcee06323b..07e5a0ed42 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java @@ -19,9 +19,12 @@ package org.springframework.integration.router; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Properties; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -50,6 +53,21 @@ import org.springframework.util.StringUtils; */ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter implements MappingMessageRouterManagement { + private static final int DEFAULT_DYNAMIC_CHANNEL_LIMIT = 100; + + private int dynamicChannelLimit = DEFAULT_DYNAMIC_CHANNEL_LIMIT; + + @SuppressWarnings("serial") + private final Map dynamicChannels = Collections.synchronizedMap( + new LinkedHashMap(DEFAULT_DYNAMIC_CHANNEL_LIMIT, 0.75f, true) { + + @Override + protected boolean removeEldestEntry(Entry eldest) { + return this.size() > AbstractMappingMessageRouter.this.dynamicChannelLimit; + } + + }); + protected volatile Map channelMappings = new ConcurrentHashMap(); private volatile String prefix; @@ -97,6 +115,18 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter this.resolutionRequired = resolutionRequired; } + /** + * Set a limit for how many dynamic channels are retained (for reporting purposes). + * When the limit is exceeded, the oldest channel is discarded. + *

NOTE: this does not affect routing, just the reporting which dynamically + * resolved channels have been routed to. Default {@code 100}. + * @param dynamicChannelLimit the limit. + * @see #getDynamicChannelNames() + */ + public void setDynamicChannelLimit(int dynamicChannelLimit) { + this.dynamicChannelLimit = dynamicChannelLimit; + } + /** * Returns an unmodifiable version of the channel mappings. * This is intended for use by subclasses only. @@ -133,6 +163,12 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter this.channelMappings = newChannelMappings; } + @Override + @ManagedAttribute + public Collection getDynamicChannelNames() { + return Collections.unmodifiableSet(this.dynamicChannels.keySet()); + } + /** * Subclasses must implement this method to return the channel keys. * A "key" might be present in this router's "channelMappings", or it @@ -208,8 +244,10 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter // if the channelMappings contains a mapping, we'll use the mapped value // otherwise, the String-based channelKey itself will be used as the channel name String channelName = channelKey; + boolean mapped = false; if (this.channelMappings.containsKey(channelKey)) { channelName = this.channelMappings.get(channelKey); + mapped = true; } if (this.prefix != null) { channelName = this.prefix + channelName; @@ -220,6 +258,9 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter MessageChannel channel = resolveChannelForName(channelName, message); if (channel != null) { channels.add(channel); + if (!mapped && !(this.dynamicChannels.get(channelName) != null)) { + this.dynamicChannels.put(channelName, channel); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 55ef98b2c0..e36cf3544f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -43,7 +43,7 @@ import org.springframework.util.Assert; */ @ManagedResource @IntegrationManagedResource -public abstract class AbstractMessageRouter extends AbstractMessageHandler { +public abstract class AbstractMessageRouter extends AbstractMessageHandler implements MessageRouter { private volatile MessageChannel defaultOutputChannel; @@ -69,6 +69,24 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { this.defaultOutputChannel = defaultOutputChannel; } + /** + * Get the default output channel. + * @return the channel. + * @since 4.3 + */ + @Override + public MessageChannel getDefaultOutputChannel() { + if (this.defaultOutputChannelName != null) { + synchronized (this) { + if (this.defaultOutputChannelName != null) { + this.defaultOutputChannel = getChannelResolver().resolveDestination(this.defaultOutputChannelName); + this.defaultOutputChannelName = null; + } + } + } + return this.defaultOutputChannel; + } + public void setDefaultOutputChannelName(String defaultOutputChannelName) { Assert.hasText(defaultOutputChannelName, "'defaultOutputChannelName' must not be empty"); this.defaultOutputChannelName = defaultOutputChannelName; @@ -188,14 +206,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { } } if (!sent) { - if (this.defaultOutputChannelName != null) { - synchronized (this) { - if (this.defaultOutputChannelName != null) { - this.defaultOutputChannel = getChannelResolver().resolveDestination(this.defaultOutputChannelName); - this.defaultOutputChannelName = null; - } - } - } + getDefaultOutputChannel(); if (this.defaultOutputChannel != null) { this.messagingTemplate.send(this.defaultOutputChannel, message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/ExpressionEvaluatingRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/ExpressionEvaluatingRouter.java index 3a4189e5f6..c20905c4b4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/ExpressionEvaluatingRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/ExpressionEvaluatingRouter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,12 +25,14 @@ import org.springframework.integration.handler.ExpressionEvaluatingMessageProces * resolved to a channel name or a Collection (or Array) of strings. * * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class ExpressionEvaluatingRouter extends AbstractMessageProcessingRouter { public ExpressionEvaluatingRouter(Expression expression) { super(new ExpressionEvaluatingMessageProcessor(expression)); + setPrimaryExpression(expression); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/MessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageRouter.java new file mode 100644 index 0000000000..04a40307c3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageRouter.java @@ -0,0 +1,36 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.router; + +import org.springframework.messaging.MessageChannel; + +/** + * Routers implementing this interface have a default output channel. + * + * @author Gary Russell + * @since 4.3 + * + */ +public interface MessageRouter { + + /** + * Get the default output channel. + * @return the channel. + */ + MessageChannel getDefaultOutputChannel(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/ExpressionEvaluatingSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/ExpressionEvaluatingSplitter.java index 2399799cca..397398e8ae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/ExpressionEvaluatingSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/ExpressionEvaluatingSplitter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -34,6 +34,7 @@ public class ExpressionEvaluatingSplitter extends AbstractMessageProcessingSplit @SuppressWarnings({"unchecked", "rawtypes"}) public ExpressionEvaluatingSplitter(Expression expression) { super(new ExpressionEvaluatingMessageProcessor(expression)); + setPrimaryExpression(expression); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/MappingMessageRouterManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MappingMessageRouterManagement.java index 33b083110c..9ad51be164 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/MappingMessageRouterManagement.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MappingMessageRouterManagement.java @@ -16,6 +16,7 @@ package org.springframework.integration.support.management; +import java.util.Collection; import java.util.Map; import java.util.Properties; @@ -53,7 +54,6 @@ public interface MappingMessageRouterManagement { /** * Provide mappings from channel keys to channel names. * @param channelMappings The channel mappings. - * * @since 4.0 */ @ManagedOperation @@ -61,7 +61,6 @@ public interface MappingMessageRouterManagement { /** * @return an unmodifiable map of channel mappings. - * * @since 4.0 */ @ManagedAttribute @@ -70,12 +69,21 @@ public interface MappingMessageRouterManagement { /** * Provide mappings from channel keys to channel names. * Channel names will be resolved by the {@link DestinationResolver}. - * * @param channelMappings The channel mappings. - * * @since 4.0 */ @ManagedAttribute void setChannelMappings(Map channelMappings); + /** + * Provide a collection of channel names to which + * we have routed messages where the channel was not explicitly mapped. + *

Implementations may choose to return only the most recent channel names. + * @return a collection of channel names to which + * we have routed messages where the channel was not explicitly mapped. + * @since 4.3 + */ + @ManagedAttribute + Collection getDynamicChannelNames(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/RouterMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/RouterMetrics.java index 56525fff8d..d20c798872 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/RouterMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/RouterMetrics.java @@ -16,6 +16,7 @@ package org.springframework.integration.support.management; +import java.util.Collection; import java.util.Map; import java.util.Properties; @@ -62,4 +63,9 @@ public class RouterMetrics extends LifecycleMessageHandlerMetrics implements Map this.router.setChannelMappings(channelMappings); } + @Override + public Collection getDynamicChannelNames() { + return this.router.getDynamicChannelNames(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableRoutingNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableRoutingNode.java new file mode 100644 index 0000000000..4a69431f89 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableRoutingNode.java @@ -0,0 +1,46 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.support.management.graph; + +import java.util.Collection; + +import org.springframework.messaging.MessageHandler; + +/** + * Represents an endpoint that can route to multiple channels and can emit errors + * (pollable endpoint). + * + * @author Gary Russell + * @since 4.3 + * + */ +public class ErrorCapableRoutingNode extends RoutingMessageHandlerNode implements ErrorCapableNode { + + private final String errors; + + public ErrorCapableRoutingNode(int nodeId, String name, MessageHandler handler, String input, String output, + String errors, Collection routes) { + super(nodeId, name, handler, input, output, routes); + this.errors = errors; + } + + @Override + public String getErrors() { + return this.errors; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java index 4d95292986..b3bbfc0a3b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java @@ -38,7 +38,10 @@ import org.springframework.integration.gateway.GatewayProxyFactoryBean; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.handler.CompositeMessageHandler; import org.springframework.integration.handler.DiscardingMessageHandler; +import org.springframework.integration.router.RecipientListRouter.Recipient; +import org.springframework.integration.router.RecipientListRouterManagement; import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.support.management.MappingMessageRouterManagement; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.StringUtils; @@ -172,8 +175,14 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat for (Entry gwEntry : methodMap.entrySet()) { MessagingGatewaySupport gateway = gwEntry.getValue(); Method method = gwEntry.getKey(); + Class[] parameterTypes = method.getParameterTypes(); + String[] parameterTypeNames = new String[parameterTypes.length]; + int i = 0; + for (Class type : parameterTypes) { + parameterTypeNames[i++] = type.getName(); + } String signature = method.getName() + - "(" + StringUtils.arrayToCommaDelimitedString(method.getParameterTypes()) + ")"; + "(" + StringUtils.arrayToCommaDelimitedString(parameterTypeNames) + ")"; MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode( entry.getKey().substring(1) + "." + signature, gateway); nodes.add(gatewayNode); @@ -232,6 +241,15 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat links.add(new LinkNode(endpointNode.getNodeId(), channelNode.getNodeId(), LinkNode.Type.discard)); } } + if (endpointNode instanceof RoutingMessageHandlerNode) { + Collection routes = ((RoutingMessageHandlerNode) endpointNode).getRoutes(); + for (String route : routes) { + channelNode = channelNodes.get(route); + if (channelNode != null) { + links.add(new LinkNode(endpointNode.getNodeId(), channelNode.getNodeId(), LinkNode.Type.route)); + } + } + } } /** @@ -275,13 +293,26 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat MessageChannel outputChannel = consumer.getOutputChannel(); String outputChannelName = outputChannel == null ? null : outputChannel.toString(); MessageHandler handler = consumer.getHandler(); - return handler instanceof CompositeMessageHandler - ? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null, false) - : handler instanceof DiscardingMessageHandler - ? discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null, - false) - : new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), outputChannelName); + if (handler instanceof CompositeMessageHandler) { + return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null, + false); + } + else if (handler instanceof DiscardingMessageHandler) { + return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null, + false); + } + else if (handler instanceof MappingMessageRouterManagement) { + return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler, + outputChannelName, null, false); + } + else if (handler instanceof RecipientListRouterManagement) { + return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler, + outputChannelName, null, false); + } + else { + return new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), outputChannelName); + } } private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) { @@ -290,14 +321,26 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat String errorChannel = consumer.getDefaultErrorChannel() != null ? consumer.getDefaultErrorChannel().toString() : null; MessageHandler handler = consumer.getHandler(); - return handler instanceof CompositeMessageHandler - ? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, errorChannel, - true) - : handler instanceof DiscardingMessageHandler - ? discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, - errorChannel, true) - : new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), outputChannelName, errorChannel); + if (handler instanceof CompositeMessageHandler) { + return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, + errorChannel, true); + } + else if (handler instanceof DiscardingMessageHandler) { + return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, + errorChannel, true); + } + else if (handler instanceof MappingMessageRouterManagement) { + return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler, + outputChannelName, errorChannel, true); + } + else if (handler instanceof RecipientListRouterManagement) { + return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler, + outputChannelName, errorChannel, true); + } + else { + return new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), outputChannelName, errorChannel); + } } private MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer, @@ -314,18 +357,48 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat } return polled ? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), output, errors, innerHandlers) + consumer.getInputChannel().toString(), output, errors, innerHandlers) : new CompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), output, innerHandlers); + consumer.getInputChannel().toString(), output, innerHandlers); } private MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer, DiscardingMessageHandler handler, String output, String errors, boolean polled) { return polled ? new ErrorCapableDiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString(), errors) + consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString(), errors) : new DiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, - consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString()); + consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString()); + } + + private MessageHandlerNode routingHandler(String name, IntegrationConsumer consumer, MessageHandler handler, + MappingMessageRouterManagement router, String output, String errors, boolean polled) { + Collection routes = router.getChannelMappings().values(); + Collection dynamicChannelNames = router.getDynamicChannelNames(); + if (dynamicChannelNames.size() > 0) { + routes = new ArrayList(routes); + routes.addAll(dynamicChannelNames); + } + return polled + ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), output, errors, routes) + : new RoutingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), output, routes); + } + + private MessageHandlerNode recipientListRoutingHandler(String name, IntegrationConsumer consumer, + MessageHandler handler, RecipientListRouterManagement router, String output, String errors, + boolean polled) { + Collection recipients = router.getRecipients(); + List routes = new ArrayList(recipients.size()); + for (Object recipient : recipients) { + routes.add(((Recipient) recipient).getChannel().toString()); + } + return polled + ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), output, errors, routes) + : new RoutingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, + consumer.getInputChannel().toString(), output, routes); } private void reset() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java index fead2b5d70..8673407b92 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java @@ -16,6 +16,11 @@ package org.springframework.integration.support.management.graph; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.expression.Expression; +import org.springframework.integration.context.ExpressionCapable; import org.springframework.integration.support.context.NamedComponent; /** @@ -35,12 +40,20 @@ public abstract class IntegrationNode { private final String componentType; + private final Map properties = new HashMap(); + protected IntegrationNode(int nodeId, String name, Object nodeObject, Stats stats) { this.nodeId = nodeId; this.name = name; this.componentType = nodeObject instanceof NamedComponent ? ((NamedComponent) nodeObject).getComponentType() : nodeObject.getClass().getSimpleName(); this.stats = stats; + if (nodeObject instanceof ExpressionCapable) { + Expression expression = ((ExpressionCapable) nodeObject).getExpression(); + if (expression != null) { + this.properties.put("expression", expression.getExpressionString()); + } + } } public int getNodeId() { @@ -59,6 +72,10 @@ public abstract class IntegrationNode { return this.stats.isAvailable() ? this.stats : null; } + public Map getProperties() { + return this.properties.size() == 0 ? null : this.properties; + } + public static class Stats { protected boolean isAvailable() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java index fa9c16f8d7..6f421bb6c4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java @@ -50,7 +50,7 @@ public class LinkNode { } public enum Type { - input, output, error, discard + input, output, error, discard, route } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/RoutingMessageHandlerNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/RoutingMessageHandlerNode.java new file mode 100644 index 0000000000..edce7a26a8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/RoutingMessageHandlerNode.java @@ -0,0 +1,44 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.support.management.graph; + +import java.util.Collection; + +import org.springframework.messaging.MessageHandler; + +/** + * Represents an endpoint that can route to multiple channels. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class RoutingMessageHandlerNode extends MessageHandlerNode { + + private final Collection routes; + + public RoutingMessageHandlerNode(int nodeId, String name, MessageHandler handler, String input, String output, + Collection routes) { + super(nodeId, name, handler, input, output); + this.routes = routes; + } + + public Collection getRoutes() { + return this.routes; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/HeaderValueRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/HeaderValueRouterTests.java index f5dc431535..b645b410fe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/HeaderValueRouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/HeaderValueRouterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,10 +16,14 @@ package org.springframework.integration.router; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.equalTo; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import org.junit.Test; + import org.springframework.beans.factory.support.ManagedMap; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.StaticApplicationContext; @@ -33,6 +37,7 @@ import org.springframework.messaging.core.DestinationResolver; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class HeaderValueRouterTests { @@ -51,6 +56,7 @@ public class HeaderValueRouterTests { Message result = testChannel.receive(1000); assertNotNull(result); assertSame(message, result); + context.close(); } @Test @@ -84,6 +90,7 @@ public class HeaderValueRouterTests { result = channel.receive(1000); assertNotNull(result); assertSame(message, result); + context.close(); } @Test @@ -107,7 +114,9 @@ public class HeaderValueRouterTests { Message result = channel.receive(1000); assertNotNull(result); assertSame(message, result); + context.close(); } + @Test @SuppressWarnings({ "unchecked", "rawtypes" }) public void resolveChannelNameFromMapAndCustomeResolver() { @@ -120,6 +129,7 @@ public class HeaderValueRouterTests { routerBeanDefinition.getPropertyValues().addPropertyValue("channelMappings", channelMappings); routerBeanDefinition.getPropertyValues().addPropertyValue("beanFactory", context); routerBeanDefinition.getPropertyValues().addPropertyValue("channelResolver", new DestinationResolver() { + @Override public MessageChannel resolveDestination(String channelName) { return context.getBean("anotherChannel", MessageChannel.class); } @@ -135,6 +145,7 @@ public class HeaderValueRouterTests { Message result = channel.receive(1000); assertNotNull(result); assertSame(message, result); + context.close(); } @Test @@ -159,6 +170,7 @@ public class HeaderValueRouterTests { assertNotNull(result2); assertSame(message, result1); assertSame(message, result2); + context.close(); } @Test @@ -183,7 +195,33 @@ public class HeaderValueRouterTests { assertNotNull(result2); assertSame(message, result1); assertSame(message, result2); + context.close(); } + @Test + public void dynamicChannelCache() { + StaticApplicationContext context = new StaticApplicationContext(); + RootBeanDefinition routerBeanDefinition = new RootBeanDefinition(HeaderValueRouter.class); + routerBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue("testHeaderName"); + routerBeanDefinition.getPropertyValues().addPropertyValue("resolutionRequired", "true"); + routerBeanDefinition.getPropertyValues().addPropertyValue("dynamicChannelLimit", "2"); + context.registerBeanDefinition("router", routerBeanDefinition); + context.registerBeanDefinition("channel1", new RootBeanDefinition(QueueChannel.class)); + context.registerBeanDefinition("channel2", new RootBeanDefinition(QueueChannel.class)); + context.registerBeanDefinition("channel3", new RootBeanDefinition(QueueChannel.class)); + context.refresh(); + MessageHandler handler = (MessageHandler) context.getBean("router"); + String channels = "channel1, channel2, channel1, channel3"; + Message message = MessageBuilder.withPayload("test").setHeader("testHeaderName", channels).build(); + handler.handleMessage(message); + QueueChannel channel1 = (QueueChannel) context.getBean("channel1"); + QueueChannel channel2 = (QueueChannel) context.getBean("channel2"); + QueueChannel channel3 = (QueueChannel) context.getBean("channel3"); + assertThat(channel1.getQueueSize(), equalTo(2)); + assertThat(channel2.getQueueSize(), equalTo(1)); + assertThat(channel3.getQueueSize(), equalTo(1)); + assertThat(context.getBean(HeaderValueRouter.class).getDynamicChannelNames(), contains("channel1", "channel3")); + context.close(); + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java index ec80baf8f9..1b2e2b91a3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java @@ -22,6 +22,7 @@ import static org.hamcrest.Matchers.notNullValue; import static org.junit.Assert.assertThat; import java.io.ByteArrayOutputStream; +import java.util.Arrays; import java.util.List; import java.util.Map; @@ -32,8 +33,10 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; +import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessagePublishingErrorHandler; @@ -44,8 +47,13 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.router.ExpressionEvaluatingRouter; +import org.springframework.integration.router.HeaderValueRouter; +import org.springframework.integration.router.RecipientListRouter; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; @@ -69,6 +77,10 @@ public class IntegrationGraphServerTests { @Autowired private IntegrationGraphServer server; + @Autowired + private MessageChannel toRouter; + + @SuppressWarnings("unchecked") @Test public void test() throws Exception { Graph graph = this.server.getGraph(); @@ -76,17 +88,39 @@ public class IntegrationGraphServerTests { ObjectMapper objectMapper = new ObjectMapper(); objectMapper.enable(SerializationFeature.INDENT_OUTPUT); objectMapper.writeValue(baos, graph); + // System . out . println(new String(baos.toByteArray())); + Map map = objectMapper.readValue(baos.toByteArray(), Map.class); assertThat(map.size(), is(equalTo(3))); - @SuppressWarnings("unchecked") List> nodes = (List>) map.get("nodes"); assertThat(nodes, is(notNullValue())); - assertThat(nodes.size(), is(equalTo(22))); - @SuppressWarnings("unchecked") + assertThat(nodes.size(), is(equalTo(31))); List> links = (List>) map.get("links"); assertThat(links, is(notNullValue())); - assertThat(links.size(), is(equalTo(20))); + assertThat(links.size(), is(equalTo(32))); + + toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build()); + toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "baz").build()); + toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "quxChannel").build()); + toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "fizChannel").build()); + + this.server.rebuild(); + graph = this.server.getGraph(); + baos = new ByteArrayOutputStream(); + objectMapper.enable(SerializationFeature.INDENT_OUTPUT); + objectMapper.writeValue(baos, graph); + +// System . out . println(new String(baos.toByteArray())); + + map = objectMapper.readValue(baos.toByteArray(), Map.class); + assertThat(map.size(), is(equalTo(3))); + nodes = (List>) map.get("nodes"); + assertThat(nodes, is(notNullValue())); + assertThat(nodes.size(), is(equalTo(31))); + links = (List>) map.get("links"); + assertThat(links, is(notNullValue())); + assertThat(links.size(), is(equalTo(34))); } @Configuration @@ -165,6 +199,64 @@ public class IntegrationGraphServerTests { return poller; } + @Bean + @Router(inputChannel = "toRouter") + public HeaderValueRouter router() { + HeaderValueRouter router = new HeaderValueRouter("foo"); + router.setChannelMapping("bar", "barChannel"); + router.setChannelMapping("baz", "bazChannel"); + router.setDefaultOutputChannel(discards()); + return router; + } + + @Bean + @Router(inputChannel = "four") + public RecipientListRouter rlRouter() { + RecipientListRouter router = new RecipientListRouter(); + router.setChannels(Arrays.asList(barChannel(), bazChannel())); + router.setDefaultOutputChannel(discards()); + return router; + } + + @Bean + @Router(inputChannel = "four") + public ExpressionEvaluatingRouter expressionRouter() { + ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter( + new SpelExpressionParser().parseExpression("headers['foo']")); + router.setDefaultOutputChannel(discards()); + return router; + } + + @Bean + public MessageChannel discards() { + return new DirectChannel(); + } + + @Bean + public MessageChannel toRouter() { + return new DirectChannel(); + } + + @Bean + public MessageChannel barChannel() { + return new QueueChannel(); + } + + @Bean + public MessageChannel bazChannel() { + return new QueueChannel(); + } + + @Bean + public MessageChannel quxChannel() { + return new QueueChannel(); + } + + @Bean + public MessageChannel fizChannel() { + return new QueueChannel(); + } + } public static class Services { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 25217c4ccc..afa956d2ae 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -307,9 +307,11 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply Assert.notNull(remoteFileTemplate, "'remoteFileTemplate' cannot be null"); this.remoteFileTemplate = remoteFileTemplate; this.command = command; + Expression parsedExpression = new SpelExpressionParser().parseExpression(expression); this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor( - new SpelExpressionParser().parseExpression(expression)); + parsedExpression); this.messageSessionCallback = null; + setPrimaryExpression(parsedExpression); } /** diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 3aa2ec49af..35d10c52d9 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -217,6 +217,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp Assert.notNull(requestDestinationExpression, "'requestDestinationExpression' must not be null"); this.requestDestinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor(requestDestinationExpression); + setPrimaryExpression(requestDestinationExpression); } /** diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java index 1d80bd2dd2..6e5ef2dfdd 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java @@ -70,6 +70,7 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { Assert.isTrue(this.destination == null && this.destinationName == null, "The 'destination', 'destinationName', and 'destinationExpression' properties are mutually exclusive."); this.destinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor(destinationExpression); + setPrimaryExpression(destinationExpression); } public void setHeaderMapper(JmsHeaderMapper headerMapper) { diff --git a/src/reference/asciidoc/graph.adoc b/src/reference/asciidoc/graph.adoc index 75e1bceeec..a2a8067796 100644 --- a/src/reference/asciidoc/graph.adoc +++ b/src/reference/asciidoc/graph.adoc @@ -84,6 +84,10 @@ The possible types are: - _output_ - the direction from `MessageHandler`, `MessageProducer` or `SourcePollingChannelAdapter` to the `MessageChannel` via an `outputChannel` or `replyChannel` property; - _error_ - from `MessageHandler` on `PollingConsumer` or `MessageProducer` or `SourcePollingChannelAdapter` to the `MessageChannel` via an `errorChannel` property; - _discard_ - from `DiscardingMessageHandler` (e.g. `MessageFilter`) to the `MessageChannel` via `errorChannel` property. +- _route_ - from `AbstractMappingMessageRouter` (e.g. `HeaderValueRouter`) to the `MessageChannel`. +Similar to _output_ but determined at run-time. +May be a configured channel mapping, or a dynamically resolved channel. +Routers will typically only retain up to 100 dynamic routes for this purpose, but this can be modified using the `dynamicChannelLimit` property. The information from this element can be used by a visualizing tool to render connections between nodes from the `nodes` graph element, where the `from` and `to` numbers represent the value from the `nodeId` property of the linked nodes. For example the link `type` can be used to determine the proper _port_ on the target node: @@ -105,6 +109,8 @@ For example the link `type` can be used to determine the proper _port_ on the ta ---- The `nodes` graph element is perhaps the most interesting because its elements contain not only the runtime components with their `componentType` s and `name` s, but can also optionally contain metrics exposed by the component. +Node elements contain various properties which are generally self-explanatory. +For example, expression-based components include the `expression` property containing the primary expression string for the component. To enable the metrics, add an `@EnableIntegrationManagement` to some `@Configuration` class or add an `` element to your XML configuration. You can control exactly which components in the framework collect statistics. See <> for complete information.