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

INT-3967: Expose Object Model

Browse Source 
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
This commit is contained in:
Gary Russell
2016-04-13 17:52:47 -04:00
committed by Artem Bilan
parent 3638275783
commit ed40582121
22 changed files with 1083 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java
View File

@@ -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();
}

23
spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
View File

@@ -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() {

51
spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java Normal file
View File

@@ -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();
}

30
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -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) {

24
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -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) {

9
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -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;

1
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -106,6 +106,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
this.messagingTemplate.setDestinationResolver(getChannelResolver());
}
@Override
public MessageChannel getOutputChannel() {
if (this.outputChannelName != null) {
synchronized (this) {

3
spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java
View File

@@ -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);
}

43
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java Normal file
View File

@@ -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;
}
}

49
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/Graph.java Normal file
View File

@@ -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<IntegrationNode> nodes;
private final Collection<LinkNode> links;
public Graph(Collection<IntegrationNode> nodes, Collection<LinkNode> links) {
this.nodes = nodes;
this.links = links;
}
public Collection<IntegrationNode> getNodes() {
return this.nodes;
}
public Collection<LinkNode> getLinks() {
return this.links;
}
}

194
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java Normal file
View File

@@ -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<ContextRefreshedEvent> {
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<String, MessageChannel> channels = this.applicationContext
.getBeansOfType(MessageChannel.class);
Map<String, SourcePollingChannelAdapter> spcas = this.applicationContext
.getBeansOfType(SourcePollingChannelAdapter.class);
Map<String, MessagingGatewaySupport> gateways = this.applicationContext
.getBeansOfType(MessagingGatewaySupport.class);
Map<String, MessageProducerSupport> producers = this.applicationContext
.getBeansOfType(MessageProducerSupport.class);
Map<String, IntegrationConsumer> consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class);
Collection<IntegrationNode> nodes = new ArrayList<IntegrationNode>();
Collection<LinkNode> links = new ArrayList<LinkNode>();
Map<String, MessageChannelNode> channelNodes = new HashMap<String, MessageChannelNode>();
for (Entry<String, MessageChannel> 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<String, SourcePollingChannelAdapter> 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<String, MessagingGatewaySupport> 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<String, MessageProducerSupport> 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<String, IntegrationConsumer> 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);
}
}
}

78
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationNode.java Normal file
View File

@@ -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;
}
}
}

45
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/LinkNode.java Normal file
View File

@@ -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;
}
}

117
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageChannelNode.java Normal file
View File

@@ -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();
}
}
}

54
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java Normal file
View File

@@ -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();
}
}
}

103
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageHandlerNode.java Normal file
View File

@@ -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();
}
}
}

34
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java Normal file
View File

@@ -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());
}
}

57
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java Normal file
View File

@@ -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();
}
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes related to the runtime object graph.
*/
package org.springframework.integration.support.management.graph;

2
spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java
View File

@@ -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

5
spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java
View File

@@ -180,6 +180,11 @@ public class MessageHandlerChainTests {
}
@Override
public MessageChannel getOutputChannel() {
return this.output;
}
@Override
public void handleMessage(Message<?> message) {
messageHandler.handleMessage(message);

159
spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java Normal file
View File

@@ -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<Map<?, ?>> nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes, is(notNullValue()));
assertThat(nodes.size(), is(equalTo(12)));
@SuppressWarnings("unchecked")
List<Map<?, ?>> links = (List<Map<?, ?>>) 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
}
}
}