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 ebcb63c903..f9748a446f 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 @@ -18,12 +18,16 @@ package org.springframework.integration.graph; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; @@ -42,9 +46,9 @@ 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.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.util.StringUtils; /** * Builds the runtime object model graph. @@ -67,11 +71,17 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat private String applicationName; + private Function> additionalPropertiesCallback; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; //NOSONAR (sync) } + protected ApplicationContext getApplicationContext() { + return this.applicationContext; + } + /** * Set the application name that will appear in the 'contentDescriptor' under * the 'name' key. If not provided, the property 'spring.application.name' from @@ -82,6 +92,24 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat this.applicationName = applicationName; //NOSONAR (sync) } + /** + * Specify a callback {@link Function} to be called against each {@link NamedComponent} + * to populate additional properties to the target {@link IntegrationNode}. + * @param additionalPropertiesCallback the {@link Function} to use for properties. + * @since 5.1 + */ + public void setAdditionalPropertiesCallback( + @Nullable Function> additionalPropertiesCallback) { + this.additionalPropertiesCallback = additionalPropertiesCallback; + } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getApplicationContext().equals(this.applicationContext)) { + buildGraph(); + } + } + /** * Return the cached graph. Although the graph is cached, the data therein (stats * etc.) are dynamic. @@ -99,11 +127,27 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat return this.graph; } - @Override - public void onApplicationEvent(ContextRefreshedEvent event) { - if (event.getApplicationContext().equals(this.applicationContext)) { - buildGraph(); - } + /** + * Rebuild the graph, re-cache it, and return it. Use this method if the application + * components have changed (added or removed). + * @return the graph. + * @see #getGraph() + */ + public Graph rebuild() { + return buildGraph(); + } + + /** + * Get beans for the provided type from the application context. + * This method can be extended for some custom logic, e.g. get beans + * from the parent application context as well. + * @param type the type for beans to obtain + * @param the type for beans to obtain + * @return a {@link Map} of bean for the provided type + * @since 5.1 + */ + protected Map getBeansOfType(Class type) { + return this.applicationContext.getBeansOfType(type, true, false); } private synchronized Graph buildGraph() { @@ -135,95 +179,129 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat } private Map channels(Collection nodes) { - Map channels = this.applicationContext - .getBeansOfType(MessageChannel.class, true, false); - Map channelNodes = new HashMap<>(); - for (Entry entry : channels.entrySet()) { - MessageChannel channel = entry.getValue(); - MessageChannelNode channelNode = this.nodeFactory.channelNode(entry.getKey(), channel); - String beanName = entry.getKey(); - nodes.add(channelNode); - channelNodes.put(beanName, channelNode); - } - return channelNodes; + return getBeansOfType(MessageChannel.class) + .entrySet() + .stream() + .map(e -> { + MessageChannel messageChannel = e.getValue(); + MessageChannelNode messageChannelNode = this.nodeFactory.channelNode(e.getKey(), messageChannel); + if (messageChannel instanceof NamedComponent) { + messageChannelNode.addProperties(getAdditionalPropertiesIfAny((NamedComponent) messageChannel)); + } + return messageChannelNode; + }) + .peek(nodes::add) + .collect(Collectors.toMap(MessageChannelNode::getName, Function.identity())); } private void pollingAdapters(Collection nodes, Collection links, Map channelNodes) { - Map spcas = this.applicationContext - .getBeansOfType(SourcePollingChannelAdapter.class, true, false); - for (Entry entry : spcas.entrySet()) { - SourcePollingChannelAdapter adapter = entry.getValue(); - MessageSourceNode sourceNode = this.nodeFactory.sourceNode(entry.getKey(), adapter); - nodes.add(sourceNode); - producerLink(links, channelNodes, sourceNode); - } + getBeansOfType(SourcePollingChannelAdapter.class) + .entrySet() + .stream() + .map(e -> { + SourcePollingChannelAdapter sourceAdapter = e.getValue(); + MessageSourceNode sourceNode = this.nodeFactory.sourceNode(e.getKey(), sourceAdapter); + sourceNode.addProperties(getAdditionalPropertiesIfAny(sourceAdapter)); + return sourceNode; + }) + .peek(nodes::add) + .forEach(sourceNode -> producerLink(links, channelNodes, sourceNode)); } private void gateways(Collection nodes, Collection links, Map channelNodes) { - Map gateways = this.applicationContext - .getBeansOfType(MessagingGatewaySupport.class, true, false); - for (Entry entry : gateways.entrySet()) { - MessagingGatewaySupport gateway = entry.getValue(); - MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(entry.getKey(), gateway); - nodes.add(gatewayNode); - producerLink(links, channelNodes, gatewayNode); - } - Map gpfbs = this.applicationContext - .getBeansOfType(GatewayProxyFactoryBean.class, true, false); + getBeansOfType(MessagingGatewaySupport.class) + .entrySet() + .stream() + .map(e -> { + MessagingGatewaySupport gateway = e.getValue(); + MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(e.getKey(), gateway); + gatewayNode.addProperties(getAdditionalPropertiesIfAny(gateway)); + return gatewayNode; + }) + .peek(nodes::add) + .forEach(gatewayNode -> producerLink(links, channelNodes, gatewayNode)); + + Map gpfbs = getBeansOfType(GatewayProxyFactoryBean.class); + for (Entry entry : gpfbs.entrySet()) { - Map methodMap = entry.getValue().getGateways(); - 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(parameterTypeNames) + ")"; - MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode( - entry.getKey().substring(1) + "." + signature, gateway); - nodes.add(gatewayNode); - producerLink(links, channelNodes, gatewayNode); - } + entry.getValue() + .getGateways() + .entrySet() + .stream() + .map(e -> { + MessagingGatewaySupport gateway = e.getValue(); + Method method = e.getKey(); + + String nodeName = + entry.getKey().substring(1) + "." + + method.getName() + + "(" + + Arrays.stream(method.getParameterTypes()) + .map(Class::getName) + .collect(Collectors.joining(",")) + + ")"; + + MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(nodeName, gateway); + gatewayNode.addProperties(getAdditionalPropertiesIfAny(gateway)); + return gatewayNode; + }) + .peek(nodes::add) + .forEach(gatewayNode -> producerLink(links, channelNodes, gatewayNode)); } } private void producers(Collection nodes, Collection links, Map channelNodes) { - Map producers = this.applicationContext - .getBeansOfType(MessageProducerSupport.class, true, false); - for (Entry entry : producers.entrySet()) { - MessageProducerSupport producer = entry.getValue(); - MessageProducerNode producerNode = this.nodeFactory.producerNode(entry.getKey(), producer); - nodes.add(producerNode); - producerLink(links, channelNodes, producerNode); - } + getBeansOfType(MessageProducerSupport.class) + .entrySet() + .stream() + .map(e -> { + MessageProducerSupport producer = e.getValue(); + MessageProducerNode producerNode = this.nodeFactory.producerNode(e.getKey(), producer); + producerNode.addProperties(getAdditionalPropertiesIfAny(producer)); + return producerNode; + }) + .peek(nodes::add) + .forEach(producerNode -> producerLink(links, channelNodes, producerNode)); } private void consumers(Collection nodes, Collection links, Map channelNodes) { - Map consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class, - true, false); - for (Entry entry : consumers.entrySet()) { - IntegrationConsumer consumer = entry.getValue(); - MessageHandlerNode handlerNode = consumer instanceof PollingConsumer - ? this.nodeFactory.polledHandlerNode(entry.getKey(), (PollingConsumer) consumer) - : this.nodeFactory.handlerNode(entry.getKey(), consumer); - nodes.add(handlerNode); - MessageChannelNode channelNode = channelNodes.get(handlerNode.getInput()); - if (channelNode != null) { - links.add(new LinkNode(channelNode.getNodeId(), handlerNode.getNodeId(), LinkNode.Type.input)); - } - producerLink(links, channelNodes, handlerNode); + getBeansOfType(IntegrationConsumer.class) + .entrySet() + .stream() + .map(e -> { + IntegrationConsumer consumer = e.getValue(); + MessageHandlerNode handlerNode = + consumer instanceof PollingConsumer + ? this.nodeFactory.polledHandlerNode(e.getKey(), (PollingConsumer) consumer) + : this.nodeFactory.handlerNode(e.getKey(), consumer); + handlerNode.addProperties(getAdditionalPropertiesIfAny(consumer)); + return handlerNode; + }) + .peek(nodes::add) + .forEach(handlerNode -> { + MessageChannelNode channelNode = channelNodes.get(handlerNode.getInput()); + if (channelNode != null) { + links.add(new LinkNode(channelNode.getNodeId(), handlerNode.getNodeId(), LinkNode.Type.input)); + } + producerLink(links, channelNodes, handlerNode); + }); + } + + @Nullable + private Map getAdditionalPropertiesIfAny(NamedComponent namedComponent) { + if (this.additionalPropertiesCallback != null) { + return this.additionalPropertiesCallback.apply(namedComponent); + } + else { + return null; } } @@ -260,16 +338,6 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat } } - /** - * Rebuild the graph, re-cache it, and return it. Use this method if the application - * components have changed (added or removed). - * @return the graph. - * @see #getGraph() - */ - public Graph rebuild() { - return buildGraph(); - } - private static final class NodeFactory { private final AtomicInteger nodeId = new AtomicInteger(); @@ -363,15 +431,17 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat private MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer, CompositeMessageHandler handler, String output, String errors, boolean polled) { - List handlers = handler.getHandlers(); - List innerHandlers = new ArrayList<>(); - for (MessageHandler innerHandler : handlers) { - if (innerHandler instanceof NamedComponent) { - NamedComponent named = (NamedComponent) innerHandler; - innerHandlers.add(new CompositeMessageHandlerNode.InnerHandler(named.getComponentName(), - named.getComponentType())); - } - } + List innerHandlers = + handler.getHandlers() + .stream() + .filter(NamedComponent.class::isInstance) + .map(NamedComponent.class::cast) + .map(named -> + new CompositeMessageHandlerNode.InnerHandler( + named.getComponentName(), + named.getComponentType())) + .collect(Collectors.toList()); + String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; return polled ? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler, @@ -395,12 +465,11 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat 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); - } + Collection routes = + Stream.concat(router.getChannelMappings().values().stream(), + router.getDynamicChannelNames().stream()) + .collect(Collectors.toList()); + String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; return polled ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, @@ -413,11 +482,12 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat 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()); - } + List routes = + router.getRecipients() + .stream() + .map(recipient -> ((Recipient) recipient).getChannel().toString()) + .collect(Collectors.toList()); + String inputChannel = consumer.getInputChannel() != null ? consumer.getInputChannel().toString() : null; return polled ? new ErrorCapableRoutingNode(this.nodeId.incrementAndGet(), name, handler, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java index d4c23424c9..b7101f8929 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java @@ -16,17 +16,22 @@ package org.springframework.integration.graph; +import java.util.Collections; 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; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; /** * Base class for all nodes. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.3 * */ @@ -40,13 +45,17 @@ public abstract class IntegrationNode { private final String componentType; - private final Map properties = new HashMap(); + private final Map properties = new HashMap<>(); + + private final Map unmodifiableProperties = Collections.unmodifiableMap(this.properties); 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.componentType = + nodeObject instanceof NamedComponent + ? ((NamedComponent) nodeObject).getComponentType() + : nodeObject.getClass().getSimpleName(); this.stats = stats; if (nodeObject instanceof ExpressionCapable) { Expression expression = ((ExpressionCapable) nodeObject).getExpression(); @@ -73,7 +82,29 @@ public abstract class IntegrationNode { } public Map getProperties() { - return this.properties.size() == 0 ? null : this.properties; + return this.unmodifiableProperties; + } + + /** + * Add extra property to the node. + * @param name the name for property + * @param value the value of the property + * @since 5.1 + */ + public void addProperty(String name, Object value) { + Assert.hasText(name, "'name' must not be null"); + this.properties.put(name, value); + } + + /** + * Add extra property to the node. + * @param properties additional properties to add + * @since 5.1 + */ + public void addProperties(@Nullable Map properties) { + if (properties != null) { + this.properties.putAll(properties); + } } public static class Stats { 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 bb2288e965..0e8c48cb46 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 @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.io.ByteArrayOutputStream; import java.util.Arrays; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -27,6 +28,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; @@ -50,6 +52,7 @@ import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.graph.Graph; import org.springframework.integration.graph.IntegrationGraphServer; +import org.springframework.integration.json.JsonPathUtils; import org.springframework.integration.router.ExpressionEvaluatingRouter; import org.springframework.integration.router.HeaderValueRouter; import org.springframework.integration.router.RecipientListRouter; @@ -67,6 +70,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; +import net.minidev.json.JSONArray; /** * @author Gary Russell @@ -97,13 +101,28 @@ public class IntegrationGraphServerTests { objectMapper.enable(SerializationFeature.INDENT_OUTPUT); objectMapper.writeValue(baos, graph); -// System . out . println(new String(baos.toByteArray())); + // System . out . println(new String(baos.toByteArray())); Map map = objectMapper.readValue(baos.toByteArray(), Map.class); assertThat(map.size()).isEqualTo(3); List> nodes = (List>) map.get("nodes"); assertThat(nodes).isNotNull(); assertThat(nodes.size()).isEqualTo(32); + + JSONArray jsonArray = + JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.componentType == 'gateway')]"); + + assertThat(jsonArray.size()).isEqualTo(3); + + Map gateway1 = (Map) jsonArray.get(0); + + Map properties = (Map) gateway1.get("properties"); + + assertThat(properties).isNotNull(); + + assertThat(properties.get("auto-startup")).isEqualTo(Boolean.TRUE); + assertThat(properties.get("running")).isEqualTo(Boolean.TRUE); + List> links = (List>) map.get("links"); assertThat(links).isNotNull(); assertThat(links.size()).isEqualTo(33); @@ -119,7 +138,7 @@ public class IntegrationGraphServerTests { objectMapper.enable(SerializationFeature.INDENT_OUTPUT); objectMapper.writeValue(baos, graph); -// System . out . println(new String(baos.toByteArray())); + // System . out . println(new String(baos.toByteArray())); map = objectMapper.readValue(baos.toByteArray(), Map.class); assertThat(map.size()).isEqualTo(3); @@ -135,7 +154,8 @@ public class IntegrationGraphServerTests { public void testIncludesDynamic() { Graph graph = this.server.getGraph(); assertThat(graph.getNodes().size()).isEqualTo(32); - IntegrationFlow flow = f -> f.handle(m -> { }); + IntegrationFlow flow = f -> f.handle(m -> { + }); IntegrationFlowRegistration reg = this.flowContext.registration(flow).register(); graph = this.server.rebuild(); assertThat(graph.getNodes().size()).isEqualTo(34); @@ -155,6 +175,16 @@ public class IntegrationGraphServerTests { public IntegrationGraphServer server() { IntegrationGraphServer server = new IntegrationGraphServer(); server.setApplicationName("myAppName:1.0"); + server.setAdditionalPropertiesCallback(namedComponent -> { + Map properties = null; + if (namedComponent instanceof SmartLifecycle) { + SmartLifecycle smartLifecycle = (SmartLifecycle) namedComponent; + properties = new HashMap<>(); + properties.put("auto-startup", smartLifecycle.isAutoStartup()); + properties.put("running", smartLifecycle.isRunning()); + } + return properties; + }); return server; } diff --git a/src/reference/asciidoc/graph.adoc b/src/reference/asciidoc/graph.adoc index 3606f7e178..49b048c284 100644 --- a/src/reference/asciidoc/graph.adoc +++ b/src/reference/asciidoc/graph.adoc @@ -82,7 +82,7 @@ or from an `AbstractReplyProducingMessageHandler` to a `MessageChannel`. For convenience and to let you determine a link's purpose, the model includes the `type` attribute. The possible types are: -* `input`: Identifes the direction from `MessageChannel` to the endpoint, `inputChannel`, or `requestChannel` property +* `input`: Identifies the direction from `MessageChannel` to the endpoint, `inputChannel`, or `requestChannel` property * `output`: The direction from the `MessageHandler`, `MessageProducer`, or `SourcePollingChannelAdapter` to the `MessageChannel` through an `outputChannel` or `replyChannel` property * `error`: From `MessageHandler` on `PollingConsumer` or `MessageProducer` or `SourcePollingChannelAdapter` to the `MessageChannel` through an `errorChannel` property; * `discard`: From `DiscardingMessageHandler` (such as `MessageFilter`) to the `MessageChannel` through an `errorChannel` property. @@ -147,6 +147,25 @@ It is also used in the `links` element to represent a relationship (connection) The `input` and `output` attributes are for the `inputChannel` and `outputChannel` properties of the `AbstractEndpoint`, `MessageHandler`, `SourcePollingChannelAdapter`, or `MessageProducerSupport`. See the next section for more information. +Starting with version 5.1, the `IntegrationGraphServer` accepts a `Function> additionalPropertiesCallback` for population of additional properties on the `IntegrationNode` for a particular `NamedComponent`. +For example you can expose the `SmartLifecycle` `autoStartup` and `running` properties into the target graph: + +==== +[source,java] +---- +server.setAdditionalPropertiesCallback(namedComponent -> { + Map properties = null; + if (namedComponent instanceof SmartLifecycle) { + SmartLifecycle smartLifecycle = (SmartLifecycle) namedComponent; + properties = new HashMap<>(); + properties.put("auto-startup", smartLifecycle.isAutoStartup()); + properties.put("running", smartLifecycle.isRunning()); + } + return properties; + }); +---- +==== + ==== Graph Runtime Model Spring Integration components have various levels of complexity. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index db7f79b778..22f4a0d6cc 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -210,3 +210,9 @@ e.g. `org.springframework.integration:type=MessageChannel,` `name="input#foo.myG It is now simpler to customize the standard Micrometer meters created by the framework. See <> for more information. + +[[x51.-integration-graph]] +=== Integration Graph Customization + +It is now possible to add additional properties to the `IntegrationNode` s via `Function> additionalPropertiesCallback` on the `IntegrationGraphServer`. +See <<<>>> for more information.