From ed4058212145e6e2d3197ca0d9a929eeb71c177c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 13 Apr 2016 17:52:47 -0400 Subject: [PATCH] INT-3967: Expose Object Model JIRA: https://jira.spring.io/browse/INT-3967, https://jira.spring.io/browse/INT-3987 Based on the XD FLO object model, each node represents an endpoint or channel. The graph contains nodes and links where the links represent a connection (publish or consume) to/from a channel. From the test case... ``` { "nodes" : [ { "name" : "three", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "channel", "componentName" : "three", "nodeId" : 1 }, { "name" : "four", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "channel", "componentName" : "four", "nodeId" : 2 }, { "name" : "nullChannel", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "channel", "componentName" : "nullChannel", "nodeId" : 3 }, { "name" : "errorChannel", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "publish-subscribe-channel", "componentName" : "errorChannel", "nodeId" : 4 }, { "name" : "one", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "channel", "componentName" : "one", "nodeId" : 5 }, { "name" : "two", "output" : null, "input" : null, "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "sendCount" : 0, "sendErrorCount" : 0, "timeSinceLastSend" : 0.0, "meanSendRate" : 0.0, "meanErrorRate" : 0.0, "meanErrorRatio" : 0.0, "meanSendDuration" : 0.0, "minSendDuration" : 0.0, "maxSendDuration" : 0.0, "standardDeviationSendDuration" : 0.0, "sendDuration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "sendRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "errorRate" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 }, "available" : true }, "componentType" : "channel", "componentName" : "two", "nodeId" : 6 }, { "name" : "producer", "output" : "one", "input" : null, "stats" : null, "componentType" : "test-producer", "componentName" : "producer", "nodeId" : 7 }, { "name" : "polling", "output" : null, "input" : "four", "stats" : null, "componentType" : "unknown", "componentName" : "unknown", "nodeId" : 8 }, { "name" : "foreignMessageHandlerNoStats", "output" : null, "input" : "three", "stats" : null, "componentType" : "unknown", "componentName" : "unknown", "nodeId" : 9 }, { "name" : "_org.springframework.integration.errorLogger", "output" : null, "input" : "errorChannel", "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "handleCount" : 0, "errorCount" : 0, "meanDuration" : 0.0, "minDuration" : 0.0, "maxDuration" : 0.0, "standardDeviationDuration" : 0.0, "activeCount" : 0, "available" : true, "duration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 } }, "componentType" : "logging-channel-adapter", "componentName" : "_org.springframework.integration.errorLogger", "nodeId" : 10 }, { "name" : "services.foo.serviceActivator", "output" : "two", "input" : "one", "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "handleCount" : 0, "errorCount" : 0, "meanDuration" : 0.0, "minDuration" : 0.0, "maxDuration" : 0.0, "standardDeviationDuration" : 0.0, "activeCount" : 0, "available" : true, "duration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 } }, "componentType" : "service-activator", "componentName" : "services.foo.serviceActivator.handler", "nodeId" : 11 }, { "name" : "services.bar.serviceActivator", "output" : null, "input" : "two", "stats" : { "countsEnabled" : true, "statsEnabled" : true, "loggingEnabled" : true, "handleCount" : 0, "errorCount" : 0, "meanDuration" : 0.0, "minDuration" : 0.0, "maxDuration" : 0.0, "standardDeviationDuration" : 0.0, "activeCount" : 0, "available" : true, "duration" : { "count" : 0, "min" : 0.0, "max" : 0.0, "mean" : 0.0, "standardDeviation" : 0.0, "countLong" : 0 } }, "componentType" : "service-activator", "componentName" : "services.bar.serviceActivator.handler", "nodeId" : 12 } ], "links" : [ { "from" : 7, "to" : 5 }, { "from" : 2, "to" : 8 }, { "from" : 1, "to" : 9 }, { "from" : 4, "to" : 10 }, { "from" : 5, "to" : 11 }, { "from" : 11, "to" : 6 }, { "from" : 6, "to" : 12 } ] } ``` Polishing Only include stats if at least counts are enabled. More Polishing Polishing Polishing - PR Comments; isAvailable()->protected Make isAvailable protected so it doesn't appear in the JSON. More Polishing - PR Comments Yet More Polishing buildGraph() -> private, rebuild() returns graph --- .../integration/core/MessageProducer.java | 9 +- .../endpoint/EventDrivenConsumer.java | 23 ++- .../endpoint/IntegrationConsumer.java | 51 +++++ .../endpoint/MessageProducerSupport.java | 30 ++- .../integration/endpoint/PollingConsumer.java | 24 ++- .../endpoint/SourcePollingChannelAdapter.java | 9 + .../AbstractMessageProducingHandler.java | 1 + .../ExponentialMovingAverageRate.java | 3 + .../management/graph/EndpointNode.java | 43 ++++ .../support/management/graph/Graph.java | 49 +++++ .../graph/IntegrationGraphServer.java | 194 ++++++++++++++++++ .../management/graph/IntegrationNode.java | 78 +++++++ .../support/management/graph/LinkNode.java | 45 ++++ .../management/graph/MessageChannelNode.java | 117 +++++++++++ .../management/graph/MessageGatewayNode.java | 54 +++++ .../management/graph/MessageHandlerNode.java | 103 ++++++++++ .../management/graph/MessageProducerNode.java | 34 +++ .../management/graph/MessageSourceNode.java | 57 +++++ .../management/graph/package-info.java | 4 + .../endpoint/MessageProducerSupportTests.java | 2 +- .../handler/MessageHandlerChainTests.java | 5 + .../graph/IntegrationGraphServerTests.java | 159 ++++++++++++++ 22 files changed, 1083 insertions(+), 11 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/Graph.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageChannelNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageHandlerNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/package-info.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java index 138596b2bc..dcb9ea59be 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.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,4 +34,11 @@ public interface MessageProducer { */ void setOutputChannel(MessageChannel outputChannel); + /** + * Return the the output channel. + * @return the channel. + * @since 4.3 + */ + MessageChannel getOutputChannel(); + } 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 58c64ea5cd..3edc2fdf93 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 @@ -18,7 +18,9 @@ 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.support.context.NamedComponent; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; @@ -31,7 +33,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Gary Russell */ -public class EventDrivenConsumer extends AbstractEndpoint { +public class EventDrivenConsumer extends AbstractEndpoint implements IntegrationConsumer { private final SubscribableChannel inputChannel; @@ -46,6 +48,25 @@ public class EventDrivenConsumer extends AbstractEndpoint { this.setPhase(Integer.MIN_VALUE); } + @Override + public MessageChannel getInputChannel() { + return this.inputChannel; + } + + @Override + public MessageChannel getOutputChannel() { + if (this.handler instanceof MessageProducer) { + return ((MessageProducer) this.handler).getOutputChannel(); + } + else { + return null; + } + } + + @Override + public MessageHandler getHandler() { + return this.handler; + } @Override protected void doStart() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java new file mode 100644 index 0000000000..64170bf8ae --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java @@ -0,0 +1,51 @@ +/* + * 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.endpoint; + +import org.springframework.integration.support.context.NamedComponent; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; + +/** + * Message consumers implement this interface, the message handler within a consumer + * may or may not emit output messages. + * + * @author Gary Russell + * @since 4.3 + * + */ +public interface IntegrationConsumer extends NamedComponent { + + /** + * Return the input channel. + * @return the input channel. + */ + MessageChannel getInputChannel(); + + /** + * Return the output channel (may be null). + * @return the output channel. + */ + MessageChannel getOutputChannel(); + + /** + * Return the consumer's handler. + * @return the handler. + */ + MessageHandler getHandler(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 7f7c5fe258..e7aee71eb2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; @@ -35,7 +36,8 @@ import org.springframework.util.StringUtils; * @author Artem Bilan * @author Gary Russell */ -public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent { +public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent, + SmartInitializingSingleton { private final MessagingTemplate messagingTemplate = new MessagingTemplate(); @@ -67,6 +69,19 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements this.outputChannelName = outputChannelName; } + @Override + public MessageChannel getOutputChannel() { + if (this.outputChannelName != null) { + synchronized (this) { + if (this.outputChannelName != null) { + this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); + this.outputChannelName = null; + } + } + } + return this.outputChannel; + } + public void setErrorChannel(MessageChannel errorChannel) { this.errorChannel = errorChannel; } @@ -84,6 +99,12 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements return this.messagingTemplate; } + @Override + public void afterSingletonsInstantiated() { + Assert.state(this.outputChannel != null || StringUtils.hasText(this.outputChannelName), + "'outputChannel' or 'outputChannelName' is required"); + } + @Override protected void onInit() { if (this.getBeanFactory() != null) { @@ -97,11 +118,6 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements */ @Override protected void doStart() { - Assert.state(this.outputChannel != null || StringUtils.hasText(this.outputChannelName), - "'outputChannel' or 'outputChannelName' is required"); - if (this.outputChannelName != null) { - this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); - } } /** @@ -120,7 +136,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } try { - this.messagingTemplate.send(this.outputChannel, message); + this.messagingTemplate.send(getOutputChannel(), message); } catch (RuntimeException e) { if (this.errorChannel != null) { 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 5774a28994..1b4c2bfb29 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 @@ -23,8 +23,10 @@ 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.transaction.IntegrationResourceHolder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; @@ -43,7 +45,7 @@ import org.springframework.util.CollectionUtils; * @author Gary Russell * @author Artem Bilan */ -public class PollingConsumer extends AbstractPollingEndpoint { +public class PollingConsumer extends AbstractPollingEndpoint implements IntegrationConsumer { private final PollableChannel inputChannel; @@ -71,6 +73,26 @@ public class PollingConsumer extends AbstractPollingEndpoint { this.receiveTimeout = receiveTimeout; } + @Override + public MessageChannel getInputChannel() { + return this.inputChannel; + } + + @Override + public MessageChannel getOutputChannel() { + if (this.handler instanceof MessageProducer) { + return ((MessageProducer) this.handler).getOutputChannel(); + } + else { + return null; + } + } + + @Override + public MessageHandler getHandler() { + return this.handler; + } + @Override protected void doStart() { if (this.handler instanceof Lifecycle) { 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 03d2db67fb..25012ed580 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 @@ -77,6 +77,15 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint this.outputChannel = outputChannel; } + /** + * Return this endpoint's source. + * @return the source. + * @since 4.3 + */ + public MessageSource getMessageSource() { + return this.source; + } + public void setOutputChannelName(String outputChannelName) { Assert.hasText(outputChannelName, "'outputChannelName' must not be empty"); this.outputChannelName = outputChannelName; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index e729a7792a..d1e45a8308 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -106,6 +106,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan this.messagingTemplate.setDestinationResolver(getChannelResolver()); } + @Override public MessageChannel getOutputChannel() { if (this.outputChannelName != null) { synchronized (this) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 6a51b85b38..59a2f491c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -188,6 +188,9 @@ public class ExponentialMovingAverageRate { * @return the time in seconds since the last measurement */ public double getTimeSinceLastMeasurement() { + if (this.count == 0) { + return 0; + } double t0 = lastTime(); return (System.nanoTime() / this.factor - t0); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java new file mode 100644 index 0000000000..a02346546d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java @@ -0,0 +1,43 @@ +/* + * 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; + +/** + * Base class for all endpoints. + * + * @author Gary Russell + * @since 4.3 + * + */ +public abstract class EndpointNode extends IntegrationNode { + + private final String output; + + protected EndpointNode(int nodeId, String name, Object nodeObject, Stats stats) { + this(nodeId, name, nodeObject, null, stats); + } + + protected EndpointNode(int nodeId, String name, Object nodeObject, String output, Stats stats) { + super(nodeId, name, nodeObject, stats); + this.output = output; + } + + public String getOutput() { + return this.output; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/Graph.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/Graph.java new file mode 100644 index 0000000000..89da5c132d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/Graph.java @@ -0,0 +1,49 @@ +/* + * 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; + +/** + * This object can be exposed, for example, as a JSON object over + * HTTP. + * + * @author Andy Clement + * @author Gary Russell + * @since 4.3 + * + */ +public class Graph { + + private final Collection nodes; + + private final Collection links; + + public Graph(Collection nodes, Collection links) { + this.nodes = nodes; + this.links = links; + } + + public Collection getNodes() { + return this.nodes; + } + + public Collection getLinks() { + return this.links; + } + +} 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 new file mode 100644 index 0000000000..dd8e2666c6 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java @@ -0,0 +1,194 @@ +/* + * 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.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationListener; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.integration.endpoint.IntegrationConsumer; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.messaging.MessageChannel; + +/** + * Builds the runtime object model graph. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class IntegrationGraphServer implements ApplicationContextAware, ApplicationListener { + + private final NodeFactory nodeFactory = new NodeFactory(); + + private ApplicationContext applicationContext; + + private Graph graph; + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + /** + * Return the cached graph. Although the graph is cached, the data therein (stats + * etc.) are dynamic. + * @return the graph. + * @see #rebuild() + */ + public Graph getGraph() { + if (this.graph == null) { + synchronized (this) { + if (this.graph == null) { + buildGraph(); + } + } + } + return this.graph; + } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getApplicationContext().equals(this.applicationContext)) { + buildGraph(); + } + } + + private synchronized Graph buildGraph() { + this.nodeFactory.reset(); + Map channels = this.applicationContext + .getBeansOfType(MessageChannel.class); + Map spcas = this.applicationContext + .getBeansOfType(SourcePollingChannelAdapter.class); + Map gateways = this.applicationContext + .getBeansOfType(MessagingGatewaySupport.class); + Map producers = this.applicationContext + .getBeansOfType(MessageProducerSupport.class); + Map consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class); + Collection nodes = new ArrayList(); + Collection links = new ArrayList(); + 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); + } + for (Entry entry : spcas.entrySet()) { + SourcePollingChannelAdapter adapter = entry.getValue(); + MessageSourceNode sourceNode = this.nodeFactory.sourceNode(entry.getKey(), adapter); + nodes.add(sourceNode); + MessageChannelNode channelNode = channelNodes.get(sourceNode.getOutput()); + if (channelNode != null) { + links.add(new LinkNode(sourceNode.getNodeId(), channelNode.getNodeId())); + } + } + for (Entry entry : gateways.entrySet()) { + MessagingGatewaySupport gateway = entry.getValue(); + MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(entry.getKey(), gateway); + nodes.add(gatewayNode); + MessageChannelNode channelInfo = channelNodes.get(gatewayNode.getOutput()); + if (channelInfo != null) { + links.add(new LinkNode(gatewayNode.getNodeId(), channelInfo.getNodeId())); + } + } + for (Entry entry : producers.entrySet()) { + MessageProducerSupport producer = entry.getValue(); + MessageProducerNode producerNode = this.nodeFactory.producerNode(entry.getKey(), producer); + nodes.add(producerNode); + MessageChannelNode channelNode = channelNodes.get(producerNode.getOutput()); + if (channelNode != null) { + links.add(new LinkNode(producerNode.getNodeId(), channelNode.getNodeId())); + } + } + for (Entry entry : consumers.entrySet()) { + IntegrationConsumer consumer = entry.getValue(); + MessageHandlerNode handlerNode = 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())); + } + if (handlerNode.getOutput() != null) { + channelNode = channelNodes.get(handlerNode.getOutput()); + if (channelNode != null) { + links.add(new LinkNode(handlerNode.getNodeId(), channelNode.getNodeId())); + } + } + } + this.graph = new Graph(nodes, links); + return this.graph; + } + + /** + * 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 final static class NodeFactory { + + private final AtomicInteger nodeId = new AtomicInteger(); + + private MessageChannelNode channelNode(String name, MessageChannel channel) { + return new MessageChannelNode(this.nodeId.incrementAndGet(), name, channel); + } + + private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) { + return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, + gateway.getRequestChannel().toString()); + } + + private MessageProducerNode producerNode(String name, MessageProducerSupport producer) { + MessageChannel outputChannel = producer.getOutputChannel(); + return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer, outputChannel.toString()); + } + + private MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) { + return new MessageSourceNode(this.nodeId.incrementAndGet(), name, adapter.getMessageSource(), + adapter.getOutputChannel().toString()); + } + + private MessageHandlerNode handlerNode(String name, IntegrationConsumer consumer) { + MessageChannel outputChannel = consumer.getOutputChannel(); + String outputChannelName = outputChannel == null ? null : outputChannel.toString(); + return new MessageHandlerNode(this.nodeId.incrementAndGet(), name, consumer.getHandler(), + consumer.getInputChannel().toString(), outputChannelName); + } + + private void reset() { + this.nodeId.set(0); + } + + } + +} 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 new file mode 100644 index 0000000000..fe18bcea46 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java @@ -0,0 +1,78 @@ +/* + * 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 org.springframework.integration.support.context.NamedComponent; + +/** + * Base class for all nodes. + * + * @author Gary Russell + * @since 4.3 + * + */ +public abstract class IntegrationNode { + + private final int nodeId; + + private final String name; + + private final Stats stats; + + private final String componentType; + + private final String componentName; + + 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.componentName = nodeObject instanceof NamedComponent ? ((NamedComponent) nodeObject).getComponentName() + : nodeObject.toString(); + this.stats = stats; + } + + public int getNodeId() { + return this.nodeId; + } + + public String getName() { + return this.name; + } + + public String getComponentType() { + return this.componentType; + } + + public String getComponentName() { + return this.componentName; + } + + public Stats getStats() { + return this.stats.isAvailable() ? this.stats : null; + } + + public static class Stats { + + protected boolean isAvailable() { + return false; + } + + } + +} 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 new file mode 100644 index 0000000000..e2f6103c42 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java @@ -0,0 +1,45 @@ +/* + * 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; + +/** + * Represents a link between nodes. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class LinkNode { + + private final int from; + + private final int to; + + public LinkNode(int from, int to) { + this.from = from; + this.to = to; + } + + public int getFrom() { + return this.from; + } + + public int getTo() { + return this.to; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageChannelNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageChannelNode.java new file mode 100644 index 0000000000..8a8c1260d2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageChannelNode.java @@ -0,0 +1,117 @@ +/* + * 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 org.springframework.integration.support.management.MessageChannelMetrics; +import org.springframework.integration.support.management.Statistics; +import org.springframework.messaging.MessageChannel; + +/** + * Represents a message channel. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class MessageChannelNode extends IntegrationNode { + + public MessageChannelNode(int nodeId, String name, MessageChannel channel) { + super(nodeId, name, channel, channel instanceof MessageChannelMetrics + ? new Stats((MessageChannelMetrics) channel) : new IntegrationNode.Stats()); + } + + + public static final class Stats extends IntegrationNode.Stats { + + private final MessageChannelMetrics channel; + + private Stats(MessageChannelMetrics channel) { + this.channel = channel; + } + + @Override + protected boolean isAvailable() { + return this.channel.isCountsEnabled(); + } + + public boolean isCountsEnabled() { + return this.channel.isCountsEnabled(); + } + + public boolean isLoggingEnabled() { + return this.channel.isLoggingEnabled(); + } + + public long getSendCount() { + return this.channel.getSendCountLong(); + } + + public long getSendErrorCount() { + return this.channel.getSendErrorCountLong(); + } + + public double getTimeSinceLastSend() { + return this.channel.getTimeSinceLastSend(); + } + + public double getMeanSendRate() { + return this.channel.getMeanSendRate(); + } + + public double getMeanErrorRate() { + return this.channel.getMeanErrorRate(); + } + + public double getMeanErrorRatio() { + return this.channel.getMeanErrorRatio(); + } + + public double getMeanSendDuration() { + return this.channel.getMeanSendDuration(); + } + + public double getMinSendDuration() { + return this.channel.getMinSendDuration(); + } + + public double getMaxSendDuration() { + return this.channel.getMaxSendDuration(); + } + + public double getStandardDeviationSendDuration() { + return this.channel.getStandardDeviationSendDuration(); + } + + public Statistics getSendDuration() { + return this.channel.getSendDuration(); + } + + public Statistics getSendRate() { + return this.channel.getSendRate(); + } + + public Statistics getErrorRate() { + return this.channel.getErrorRate(); + } + + public boolean isStatsEnabled() { + return this.channel.isStatsEnabled(); + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java new file mode 100644 index 0000000000..3623b0fe41 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java @@ -0,0 +1,54 @@ +/* + * 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 org.springframework.integration.gateway.MessagingGatewaySupport; + +/** + * Represents an inbound gateway. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class MessageGatewayNode extends EndpointNode { + + public MessageGatewayNode(int nodeId, String name, MessagingGatewaySupport gateway, String output) { + super(nodeId, name, gateway, output, new Stats(gateway)); + } + + + public static final class Stats extends IntegrationNode.Stats { + + private final MessagingGatewaySupport gateway; + + private Stats(MessagingGatewaySupport gateway) { + this.gateway = gateway; + } + + @Override + protected boolean isAvailable() { + return this.gateway.isCountsEnabled(); + } + + public long getSendCount() { + return this.gateway.getMessageCountLong(); + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageHandlerNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageHandlerNode.java new file mode 100644 index 0000000000..4a07e947fc --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageHandlerNode.java @@ -0,0 +1,103 @@ +/* + * 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 org.springframework.integration.support.management.MessageHandlerMetrics; +import org.springframework.integration.support.management.Statistics; +import org.springframework.messaging.MessageHandler; + +/** + * Represents a message handler. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class MessageHandlerNode extends EndpointNode { + + private final String input; + + public MessageHandlerNode(int nodeId, String name, MessageHandler handler, String input, String output) { + super(nodeId, name, handler, output, handler instanceof MessageHandlerMetrics + ? new Stats((MessageHandlerMetrics) handler) : new IntegrationNode.Stats()); + this.input = input; + } + + public String getInput() { + return this.input; + } + + public static final class Stats extends IntegrationNode.Stats { + + private final MessageHandlerMetrics handler; + + private Stats(MessageHandlerMetrics handler) { + this.handler = handler; + } + + @Override + protected boolean isAvailable() { + return this.handler.isCountsEnabled(); + } + + public boolean isLoggingEnabled() { + return this.handler.isLoggingEnabled(); + } + + public long getHandleCount() { + return this.handler.getHandleCountLong(); + } + + public long getErrorCount() { + return this.handler.getErrorCountLong(); + } + + public double getMeanDuration() { + return this.handler.getMeanDuration(); + } + + public double getMinDuration() { + return this.handler.getMinDuration(); + } + + public double getMaxDuration() { + return this.handler.getMaxDuration(); + } + + public double getStandardDeviationDuration() { + return this.handler.getStandardDeviationDuration(); + } + + public long getActiveCount() { + return this.handler.getActiveCountLong(); + } + + public Statistics getDuration() { + return this.handler.getDuration(); + } + + public boolean isStatsEnabled() { + return this.handler.isStatsEnabled(); + } + + public boolean isCountsEnabled() { + return this.handler.isCountsEnabled(); + } + } + +} + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java new file mode 100644 index 0000000000..acf5db3890 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java @@ -0,0 +1,34 @@ +/* + * 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 org.springframework.integration.endpoint.MessageProducerSupport; + +/** + * Represents an inbound message producer. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class MessageProducerNode extends EndpointNode { + + public MessageProducerNode(int nodeId, String name, MessageProducerSupport producer, String output) { + super(nodeId, name, producer, output, new IntegrationNode.Stats()); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java new file mode 100644 index 0000000000..9156bd7501 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java @@ -0,0 +1,57 @@ +/* + * 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 org.springframework.integration.core.MessageSource; +import org.springframework.integration.support.management.MessageSourceMetrics; + +/** + * Represents a message source. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class MessageSourceNode extends EndpointNode { + + public MessageSourceNode(int nodeId, String name, MessageSource messageSource, String output) { + super(nodeId, name, messageSource, output, messageSource instanceof MessageSourceMetrics + ? new Stats((MessageSourceMetrics) messageSource) : new IntegrationNode.Stats()); + } + + + public static final class Stats extends IntegrationNode.Stats { + + private final MessageSourceMetrics source; + + private Stats(MessageSourceMetrics source) { + this.source = source; + } + + @Override + protected boolean isAvailable() { + return this.source.isCountsEnabled(); + } + + public long getMessageCount() { + return this.source.getMessageCountLong(); + } + + } + +} + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/package-info.java new file mode 100644 index 0000000000..a2e0dc31be --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes related to the runtime object graph. + */ +package org.springframework.integration.support.management.graph; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java index 8bc4545b12..20143f9f93 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java @@ -133,7 +133,7 @@ public class MessageProducerSupportTests { mps.setBeanFactory(testApplicationContext); mps.afterPropertiesSet(); mps.start(); - assertSame(outChannel, TestUtils.getPropertyValue(mps, "outputChannel")); + assertSame(outChannel, mps.getOutputChannel()); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java index 6f986e2fd6..335b676076 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java @@ -180,6 +180,11 @@ public class MessageHandlerChainTests { } + @Override + public MessageChannel getOutputChannel() { + return this.output; + } + @Override public void handleMessage(Message message) { messageHandler.handleMessage(message); 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 new file mode 100644 index 0000000000..0ac4d6b669 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java @@ -0,0 +1,159 @@ +/* + * 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 static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.junit.Assert.assertThat; + +import java.io.ByteArrayOutputStream; +import java.util.List; +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.EnableIntegrationManagement; +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.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + +/** + * @author Gary Russell + * @since 4.3 + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +public class IntegrationGraphServerTests { + + @Autowired + private IntegrationGraphServer server; + + @Test + public void test() throws Exception { + Graph graph = this.server.getGraph(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + 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(2))); + @SuppressWarnings("unchecked") + List> nodes = (List>) map.get("nodes"); + assertThat(nodes, is(notNullValue())); + assertThat(nodes.size(), is(equalTo(12))); + @SuppressWarnings("unchecked") + List> links = (List>) map.get("links"); + assertThat(links, is(notNullValue())); + assertThat(links.size(), is(equalTo(7))); + } + + @Configuration + @EnableIntegration + @EnableIntegrationManagement + public static class Config { + + @Bean + public IntegrationGraphServer builder() { + return new IntegrationGraphServer(); + } + + @Bean + public MessageProducer producer() { + MessageProducerSupport producer = new MessageProducerSupport() { + + @Override + public String getComponentType() { + return "test-producer"; + } + + }; + producer.setOutputChannelName("one"); + return producer; + } + + @Bean + public Services services() { + return new Services(); + } + + @Bean + public EventDrivenConsumer foreignMessageHandlerNoStats() { + return new EventDrivenConsumer(three(), new BareHandler()); + } + + @Bean + public PollingConsumer polling() { + PollingConsumer pollingConsumer = new PollingConsumer(four(), new BareHandler()); + pollingConsumer.setAutoStartup(false); + return pollingConsumer; + } + + @Bean + public SubscribableChannel three() { + return new DirectChannel(); + } + + @Bean + public PollableChannel four() { + return new QueueChannel(); + } + + } + + public static class Services { + + @ServiceActivator(inputChannel = "one", outputChannel = "two") + public String foo(String foo) { + return foo.toUpperCase(); + } + + @ServiceActivator(inputChannel = "two") + public void bar(String foo) { + } + + } + + public static class BareHandler implements MessageHandler { + + @Override + public void handleMessage(Message message) throws MessagingException { + // empty + } + + } + +}