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

INT-4038: Add GPFB Gateways to Graph

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

General format for node name: bean.method#n

The #n is needed to disambiguate methods with the same name.

Polishing according PR comments

* Fix generic type for `Collections.unmodifiableMap` usage
* Change the gateway method logic in the `IntegrationGraphServer` to include the method signature in the node name to distinguish them as unique
* Move `MutableMessageBuilderFactoryTests` into the separate nested `mutable` package since `@IntegrationComponentScan` sees a new `@MessagingGateway` in the `IntegrationGraphServerTests`.
See https://jira.spring.io/browse/INT-4040
* Document the `@MessagingGateway` representation in the `graph.adoc`
This commit is contained in:
Gary Russell
2016-05-25 15:42:27 -04:00
committed by Artem Bilan
parent 16f9c9ae49
commit 43b83bec53
5 changed files with 116 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -20,6 +20,7 @@ import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
@@ -279,6 +280,16 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
return this.asyncExecutor;
}
/**
* Return the Map of {@link Method} to {@link MessagingGatewaySupport}
* generated by this factory bean.
* @return the map.
* @since 4.3
*/
public Map<Method, MessagingGatewaySupport> getGateways() {
return Collections.<Method, MessagingGatewaySupport>unmodifiableMap(this.gatewayMap);
}
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
@@ -520,8 +531,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
if (StringUtils.hasText(payloadExpression)) {
messageMapper.setPayloadExpression(payloadExpression);
}
messageMapper.setBeanFactory(this.getBeanFactory());
MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
messageMapper.setBeanFactory(this.getBeanFactory());
MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
gateway.setErrorChannel(this.errorChannel);
if (this.getTaskScheduler() != null) {
gateway.setTaskScheduler(this.getTaskScheduler());
@@ -594,7 +605,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) {
if (!runningOnCallerThread &&
(Future.class.isAssignableFrom(method.getReturnType())
|| (reactorPresent && Promise.class.isAssignableFrom(method.getReturnType())))) {
|| (reactorPresent && Promise.class.isAssignableFrom(method.getReturnType())))) {
Type returnType = method.getGenericReturnType();
if (returnType instanceof ParameterizedType) {
Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments();

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

@@ -16,6 +16,7 @@
package org.springframework.integration.support.management.graph;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -33,12 +34,14 @@ import org.springframework.integration.endpoint.IntegrationConsumer;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.handler.CompositeMessageHandler;
import org.springframework.integration.handler.DiscardingMessageHandler;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StringUtils;
/**
* Builds the runtime object model graph.
@@ -162,6 +165,21 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
nodes.add(gatewayNode);
producerLink(links, channelNodes, gatewayNode);
}
Map<String, GatewayProxyFactoryBean> gpfbs = this.applicationContext
.getBeansOfType(GatewayProxyFactoryBean.class);
for (Entry<String, GatewayProxyFactoryBean> entry : gpfbs.entrySet()) {
Map<Method, MessagingGatewaySupport> methodMap = entry.getValue().getGateways();
for (Entry<Method, MessagingGatewaySupport> gwEntry : methodMap.entrySet()) {
MessagingGatewaySupport gateway = gwEntry.getValue();
Method method = gwEntry.getKey();
String signature = method.getName() +
"(" + StringUtils.arrayToCommaDelimitedString(method.getParameterTypes()) + ")";
MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(
entry.getKey().substring(1) + "." + signature, gateway);
nodes.add(gatewayNode);
producerLink(links, channelNodes, gatewayNode);
}
}
}
private void producers(Collection<IntegrationNode> nodes, Collection<LinkNode> links,
@@ -258,12 +276,12 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
String outputChannelName = outputChannel == null ? null : outputChannel.toString();
MessageHandler handler = consumer.getHandler();
return handler instanceof CompositeMessageHandler
? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null, false)
: handler instanceof DiscardingMessageHandler
? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null, false)
: handler instanceof DiscardingMessageHandler
? discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null,
false)
false)
: new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), outputChannelName);
consumer.getInputChannel().toString(), outputChannelName);
}
private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) {
@@ -273,13 +291,13 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
? consumer.getDefaultErrorChannel().toString() : null;
MessageHandler handler = consumer.getHandler();
return handler instanceof CompositeMessageHandler
? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, errorChannel,
true)
: handler instanceof DiscardingMessageHandler
? compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, errorChannel,
true)
: handler instanceof DiscardingMessageHandler
? discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName,
errorChannel, true)
errorChannel, true)
: new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), outputChannelName, errorChannel);
consumer.getInputChannel().toString(), outputChannelName, errorChannel);
}
private MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer,
@@ -295,19 +313,19 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
}
}
return polled
? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, errors, innerHandlers)
: new CompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, innerHandlers);
? new ErrorCapableCompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, errors, innerHandlers)
: new CompositeMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, innerHandlers);
}
private MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer,
DiscardingMessageHandler handler, String output, String errors, boolean polled) {
return polled
? new ErrorCapableDiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString(), errors)
: new DiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString());
? new ErrorCapableDiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString(), errors)
: new DiscardingMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
consumer.getInputChannel().toString(), output, handler.getDiscardChannel().toString());
}
private void reset() {

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

@@ -32,6 +32,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
@@ -80,16 +82,17 @@ public class IntegrationGraphServerTests {
@SuppressWarnings("unchecked")
List<Map<?, ?>> nodes = (List<Map<?, ?>>) map.get("nodes");
assertThat(nodes, is(notNullValue()));
assertThat(nodes.size(), is(equalTo(19)));
assertThat(nodes.size(), is(equalTo(22)));
@SuppressWarnings("unchecked")
List<Map<?, ?>> links = (List<Map<?, ?>>) map.get("links");
assertThat(links, is(notNullValue()));
assertThat(links.size(), is(equalTo(17)));
assertThat(links.size(), is(equalTo(20)));
}
@Configuration
@EnableIntegration
@EnableIntegrationManagement
@IntegrationComponentScan
@ImportResource("org/springframework/integration/support/management/graph/integration-graph-context.xml")
public static class Config {
@@ -186,4 +189,15 @@ public class IntegrationGraphServerTests {
}
@MessagingGateway(defaultRequestChannel = "four")
public interface Gate {
void foo(String foo);
void foo(Integer foo);
void bar(String bar);
}
}

5
spring-integration-core/src/test/java/org/springframework/integration/support/MutableMessageBuilderFactoryTests.java → spring-integration-core/src/test/java/org/springframework/integration/support/mutable/MutableMessageBuilderFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.support;
package org.springframework.integration.support.mutable;
import static org.junit.Assert.assertTrue;
@@ -35,6 +35,7 @@ import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.support.MutableMessageBuilderFactory;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.test.context.ContextConfiguration;