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

INT-4514: Integration Graph, include dynamic flows

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4514

`getBeansOfType()` with eager init does not currently return late registered singletons.
We should not be eagerly initializing here anyway.

**cherry-pick to 5.0.x**

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java
This commit is contained in:
Gary Russell
2018-07-27 12:21:31 -04:00
committed by Artem Bilan
parent b47c0de629
commit c50a135f23
2 changed files with 37 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -135,7 +135,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private Map<String, MessageChannelNode> channels(Collection<IntegrationNode> nodes) {
Map<String, MessageChannel> channels = this.applicationContext
.getBeansOfType(MessageChannel.class);
.getBeansOfType(MessageChannel.class, true, false);
Map<String, MessageChannelNode> channelNodes = new HashMap<String, MessageChannelNode>();
for (Entry<String, MessageChannel> entry : channels.entrySet()) {
MessageChannel channel = entry.getValue();
@@ -150,7 +150,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private void pollingAdapters(Collection<IntegrationNode> nodes, Collection<LinkNode> links,
Map<String, MessageChannelNode> channelNodes) {
Map<String, SourcePollingChannelAdapter> spcas = this.applicationContext
.getBeansOfType(SourcePollingChannelAdapter.class);
.getBeansOfType(SourcePollingChannelAdapter.class, true, false);
for (Entry<String, SourcePollingChannelAdapter> entry : spcas.entrySet()) {
SourcePollingChannelAdapter adapter = entry.getValue();
MessageSourceNode sourceNode = this.nodeFactory.sourceNode(entry.getKey(), adapter);
@@ -162,7 +162,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private void gateways(Collection<IntegrationNode> nodes, Collection<LinkNode> links,
Map<String, MessageChannelNode> channelNodes) {
Map<String, MessagingGatewaySupport> gateways = this.applicationContext
.getBeansOfType(MessagingGatewaySupport.class);
.getBeansOfType(MessagingGatewaySupport.class, true, false);
for (Entry<String, MessagingGatewaySupport> entry : gateways.entrySet()) {
MessagingGatewaySupport gateway = entry.getValue();
MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(entry.getKey(), gateway);
@@ -170,7 +170,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
producerLink(links, channelNodes, gatewayNode);
}
Map<String, GatewayProxyFactoryBean> gpfbs = this.applicationContext
.getBeansOfType(GatewayProxyFactoryBean.class);
.getBeansOfType(GatewayProxyFactoryBean.class, true, false);
for (Entry<String, GatewayProxyFactoryBean> entry : gpfbs.entrySet()) {
Map<Method, MessagingGatewaySupport> methodMap = entry.getValue().getGateways();
for (Entry<Method, MessagingGatewaySupport> gwEntry : methodMap.entrySet()) {
@@ -195,7 +195,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private void producers(Collection<IntegrationNode> nodes, Collection<LinkNode> links,
Map<String, MessageChannelNode> channelNodes) {
Map<String, MessageProducerSupport> producers = this.applicationContext
.getBeansOfType(MessageProducerSupport.class);
.getBeansOfType(MessageProducerSupport.class, true, false);
for (Entry<String, MessageProducerSupport> entry : producers.entrySet()) {
MessageProducerSupport producer = entry.getValue();
MessageProducerNode producerNode = this.nodeFactory.producerNode(entry.getKey(), producer);
@@ -206,7 +206,8 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private void consumers(Collection<IntegrationNode> nodes, Collection<LinkNode> links,
Map<String, MessageChannelNode> channelNodes) {
Map<String, IntegrationConsumer> consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class);
Map<String, IntegrationConsumer> consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class,
true, false);
for (Entry<String, IntegrationConsumer> entry : consumers.entrySet()) {
IntegrationConsumer consumer = entry.getValue();
MessageHandlerNode handlerNode = consumer instanceof PollingConsumer

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

@@ -16,10 +16,7 @@
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 static org.assertj.core.api.Assertions.assertThat;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;
@@ -45,6 +42,9 @@ 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.dsl.IntegrationFlow;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.dsl.context.IntegrationFlowRegistration;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.endpoint.PollingConsumer;
@@ -82,6 +82,9 @@ public class IntegrationGraphServerTests {
@Autowired
private MessageChannel toRouter;
@Autowired
private IntegrationFlowContext flowContext;
@SuppressWarnings("unchecked")
@Test
public void test() throws Exception {
@@ -94,13 +97,13 @@ public class IntegrationGraphServerTests {
// System . out . println(new String(baos.toByteArray()));
Map<?, ?> map = objectMapper.readValue(baos.toByteArray(), Map.class);
assertThat(map.size(), is(equalTo(3)));
assertThat(map.size()).isEqualTo(3);
List<Map<?, ?>> nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes, is(notNullValue()));
assertThat(nodes.size(), is(equalTo(32)));
assertThat(nodes).isNotNull();
assertThat(nodes.size()).isEqualTo(32);
List<Map<?, ?>> links = (List<Map<?, ?>>) map.get("links");
assertThat(links, is(notNullValue()));
assertThat(links.size(), is(equalTo(33)));
assertThat(links).isNotNull();
assertThat(links.size()).isEqualTo(33);
toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build());
toRouter.send(MessageBuilder.withPayload("foo").setHeader("foo", "baz").build());
@@ -116,13 +119,26 @@ public class IntegrationGraphServerTests {
// System . out . println(new String(baos.toByteArray()));
map = objectMapper.readValue(baos.toByteArray(), Map.class);
assertThat(map.size(), is(equalTo(3)));
assertThat(map.size()).isEqualTo(3);
nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes, is(notNullValue()));
assertThat(nodes.size(), is(equalTo(32)));
assertThat(nodes).isNotNull();
assertThat(nodes.size()).isEqualTo(32);
links = (List<Map<?, ?>>) map.get("links");
assertThat(links, is(notNullValue()));
assertThat(links.size(), is(equalTo(35)));
assertThat(links).isNotNull();
assertThat(links.size()).isEqualTo(35);
}
@Test
public void testIncludesDynamic() {
Graph graph = this.server.getGraph();
assertThat(graph.getNodes().size()).isEqualTo(32);
IntegrationFlow flow = f -> f.handle(m -> { });
IntegrationFlowRegistration reg = this.flowContext.registration(flow).register();
graph = this.server.rebuild();
assertThat(graph.getNodes().size()).isEqualTo(34);
this.flowContext.remove(reg.getId());
graph = this.server.rebuild();
assertThat(graph.getNodes().size()).isEqualTo(32);
}
@Configuration