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

INT-4265: Add Java DSL .routeByError() EIP-Method

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

* Add `.routeByError()` EIP-method to the `IntegrationFlowDefinition`
based on the `ErrorMessageExceptionTypeRouter`
* Add missed `dynamicChannelLimit()` option to the `RouterSpec`
* Rework `RouterSpec#RouterMappingProvider` into the `ContextRefreshedEvent`
phase initialization to be sure that dependent `MappingMessageRouterManagement`
is fully initialized before applying mapping conversions

* Rename `routeByError()` to `routeByException()` to more reflect reality
of the `ErrorMessageExceptionTypeRouter` and don't mislead about
Java's `Error`
* Polish `ErrorMessageExceptionTypeRouter` JavaDocs a bit and some Java 8
code style
* Add `ApplicationContext` assertion into the
`RouterMappingProvider.onApplicationEvent()` do not trigger delegate
initialization if `ContextRefreshedEvent` is from the different app context
This commit is contained in:
Artem Bilan
2017-04-26 19:10:23 -04:00
committed by Gary Russell
parent ca231763a3
commit f03356ccbc
4 changed files with 152 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -69,6 +69,7 @@ import org.springframework.integration.handler.MessageTriggerAction;
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.router.ErrorMessageExceptionTypeRouter;
import org.springframework.integration.router.ExpressionEvaluatingRouter;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.integration.router.RecipientListRouter;
@@ -2009,12 +2010,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
/**
* Populate the {@link RecipientListRouter} options from {@link RecipientListRouterSpec}.
* Populate the {@link RecipientListRouter} with options from the {@link RecipientListRouterSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .routeToRecipients(r -> r
*.recipient("bar-channel", m ->
* .recipient("bar-channel", m ->
* m.getHeaders().containsKey("recipient") && (boolean) m.getHeaders().get("recipient"))
* .recipientFlow("'foo' == payload or 'bar' == payload or 'baz' == payload",
* f -> f.transform(String.class, p -> p.toUpperCase())
@@ -2028,6 +2029,27 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return route(new RecipientListRouterSpec(), routerConfigurer);
}
/**
* Populate the {@link ErrorMessageExceptionTypeRouter} with options from the {@link RouterSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .routeByException(r -> r
* .channelMapping(IllegalArgumentException.class, "illegalArgumentChannel")
* .subFlowMapping(MessageHandlingException.class, sf ->
* sf.handle(...))
* )
* }
* </pre>
* @param routerConfigurer the {@link Consumer} to provide {@link ErrorMessageExceptionTypeRouter} options.
* @return the current {@link IntegrationFlowDefinition}.
* @see ErrorMessageExceptionTypeRouter
*/
public B routeByException(
Consumer<RouterSpec<Class<? extends Throwable>, ErrorMessageExceptionTypeRouter>> routerConfigurer) {
return route(new RouterSpec<>(new ErrorMessageExceptionTypeRouter()), routerConfigurer);
}
/**
* Populate the provided {@link AbstractMessageRouter} implementation to the
* current integration flow position.
@@ -2561,6 +2583,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return new PublisherIntegrationFlow<>(this.integrationComponents, publisher);
}
@SuppressWarnings("unchecked")
private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
Consumer<S> endpointConfigurer) {
if (endpointConfigurer != null) {

65
spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java
View File

@@ -19,7 +19,10 @@ package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.channel.DirectChannel;
@@ -68,6 +71,20 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
return _this();
}
/**
* Set a limit for how many dynamic channels are retained (for reporting purposes).
* When the limit is exceeded, the oldest channel is discarded.
* <p><b>NOTE: this does not affect routing, just the reporting which dynamically
* resolved channels have been routed to.</b> Default {@code 100}.
* @param dynamicChannelLimit the limit.
* @return the router spec.
* @see AbstractMappingMessageRouter#setDynamicChannelLimit(int)
*/
public RouterSpec<K, R> dynamicChannelLimit(int dynamicChannelLimit) {
this.handler.setDynamicChannelLimit(dynamicChannelLimit);
return _this();
}
/**
* Cannot be invoked if {@link #subFlowMapping(Object, IntegrationFlow)} is used.
* @param prefix the prefix.
@@ -163,7 +180,10 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
return super.getComponentsToRegister();
}
private static class RouterMappingProvider extends IntegrationObjectSupport {
private static class RouterMappingProvider extends IntegrationObjectSupport
implements ApplicationListener<ContextRefreshedEvent> {
private final AtomicBoolean initialized = new AtomicBoolean();
private final MappingMessageRouterManagement router;
@@ -178,28 +198,31 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
}
@Override
protected void onInit() throws Exception {
ConversionService conversionService = getConversionService();
if (conversionService == null) {
conversionService = DefaultConversionService.getSharedInstance();
}
for (Map.Entry<Object, NamedComponent> entry : this.mapping.entrySet()) {
Object key = entry.getKey();
String channelKey;
if (key instanceof String) {
channelKey = (String) key;
}
else if (key instanceof Class) {
channelKey = ((Class<?>) key).getName();
}
else if (conversionService.canConvert(key.getClass(), String.class)) {
channelKey = conversionService.convert(key, String.class);
}
else {
throw new MessagingException("unsupported channel mapping type for router [" + key.getClass() + "]");
public void onApplicationEvent(ContextRefreshedEvent event) {
if (event.getApplicationContext() == getApplicationContext() && !this.initialized.getAndSet(true)) {
ConversionService conversionService = getConversionService();
if (conversionService == null) {
conversionService = DefaultConversionService.getSharedInstance();
}
for (Map.Entry<Object, NamedComponent> entry : this.mapping.entrySet()) {
Object key = entry.getKey();
String channelKey;
if (key instanceof String) {
channelKey = (String) key;
}
else if (key instanceof Class) {
channelKey = ((Class<?>) key).getName();
}
else if (conversionService.canConvert(key.getClass(), String.class)) {
channelKey = conversionService.convert(key, String.class);
}
else {
throw new MessagingException("Unsupported channel mapping type for router ["
+ key.getClass() + "]");
}
this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
}
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -31,8 +31,9 @@ import org.springframework.util.ClassUtils;
/**
* A Message Router that resolves the target {@link MessageChannel} for
* messages whose payload is an Exception. The channel resolution is based upon
* the most specific cause of the error for which a channel-mapping exists.
* messages whose payload is a {@link Throwable}.
* The channel resolution is based upon the most specific cause
* of the error for which a channel-mapping exists.
* <p>
* The channel-mapping can be specified for the super classes to avoid mapping duplication
* for the particular exception implementation.
@@ -43,7 +44,7 @@ import org.springframework.util.ClassUtils;
*/
public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRouter {
private volatile Map<String, Class<?>> classNameMappings = new ConcurrentHashMap<String, Class<?>>();
private volatile Map<String, Class<?>> classNameMappings = new ConcurrentHashMap<>();
private volatile boolean initialized;
@@ -57,7 +58,7 @@ public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRoute
}
private void populateClassNameMapping(Set<String> classNames) {
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<String, Class<?>>();
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<>();
for (String className : classNames) {
newClassNameMappings.put(className, resolveClassFromName(className));
}
@@ -77,7 +78,7 @@ public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRoute
@ManagedOperation
public void setChannelMapping(String key, String channelName) {
super.setChannelMapping(key, channelName);
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<String, Class<?>>(this.classNameMappings);
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<>(this.classNameMappings);
newClassNameMappings.put(key, resolveClassFromName(key));
this.classNameMappings = newClassNameMappings;
}
@@ -86,7 +87,7 @@ public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRoute
@ManagedOperation
public void removeChannelMapping(String key) {
super.removeChannelMapping(key);
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<String, Class<?>>(this.classNameMappings);
Map<String, Class<?>> newClassNameMappings = new ConcurrentHashMap<>(this.classNameMappings);
newClassNameMappings.remove(key);
this.classNameMappings = newClassNameMappings;
}
@@ -122,7 +123,7 @@ public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRoute
cause = cause.getCause();
}
}
return Collections.<Object>singletonList(mostSpecificCause);
return Collections.singletonList(mostSpecificCause);
}
}

84
spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -50,14 +50,15 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author Artem Bilan
@@ -65,8 +66,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*
* @since 5.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@RunWith(SpringRunner.class)
@DirtiesContext
public class RouterTests {
@@ -90,7 +90,7 @@ public class RouterTests {
public void testRouter() {
this.beanFactory.containsBean("routeFlow.subFlow#0.channel#0");
int[] payloads = new int[] {1, 2, 3, 4, 5, 6};
int[] payloads = new int[] { 1, 2, 3, 4, 5, 6 };
for (int payload : payloads) {
this.routerInput.send(new GenericMessage<>(payload));
@@ -125,7 +125,7 @@ public class RouterTests {
@SuppressWarnings("unchecked")
List<Integer> results = (List<Integer>) payload;
assertArrayEquals(new Integer[] {3, 4, 9, 8, 15, 12}, results.toArray(new Integer[results.size()]));
assertArrayEquals(new Integer[] { 3, 4, 9, 8, 15, 12 }, results.toArray(new Integer[results.size()]));
}
@Autowired
@@ -498,6 +498,39 @@ public class RouterTests {
assertThat(((List<?>) payload).size(), greaterThanOrEqualTo(1));
}
@Autowired
@Qualifier("exceptionTypeRouteFlow.input")
private MessageChannel exceptionTypeRouteFlowInput;
@Autowired
private PollableChannel illegalArgumentChannel;
@Autowired
private PollableChannel runtimeExceptionChannel;
@Autowired
private PollableChannel messageHandlingExceptionChannel;
@Autowired
private PollableChannel exceptionRouterDefaultChannel;
@Test
public void testExceptionTypeRouteFlow() {
Message<?> failedMessage = new GenericMessage<>("foo");
IllegalArgumentException rootCause = new IllegalArgumentException("bad argument");
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
this.exceptionTypeRouteFlowInput.send(message);
assertNotNull(this.illegalArgumentChannel.receive(1000));
assertNull(this.exceptionRouterDefaultChannel.receive(0));
assertNull(this.runtimeExceptionChannel.receive(0));
assertNull(this.messageHandlingExceptionChannel.receive(0));
}
@Configuration
@EnableIntegration
@EnableMessageHistory({ "recipientListOrder*", "recipient1*", "recipient2*" })
@@ -525,8 +558,8 @@ public class RouterTests {
return f -> f
.<Boolean>route("true", m -> m
.subFlowMapping(true, sf -> sf
.<String>handle((p, h) -> p.toUpperCase())
)
.<String>handle((p, h) -> p.toUpperCase())
)
);
}
@@ -616,7 +649,7 @@ public class RouterTests {
@Bean
public IntegrationFlow routeMultiMethodInvocationFlow() {
return IntegrationFlows.from("routerMultiInput")
.route(String.class, p -> p.equals("foo") || p.equals("bar") ? new String[] {"foo", "bar"} : null,
.route(String.class, p -> p.equals("foo") || p.equals("bar") ? new String[] { "foo", "bar" } : null,
s -> s.suffix("-channel"))
.get();
}
@@ -639,6 +672,37 @@ public class RouterTests {
.channelMapping(Integer.class, "integersChannel"));
}
@Bean
public IntegrationFlow exceptionTypeRouteFlow() {
return f -> f
.routeByException(r -> r
.channelMapping(IllegalArgumentException.class, "illegalArgumentChannel")
.channelMapping(RuntimeException.class, "runtimeExceptionChannel")
.subFlowMapping(MessageHandlingException.class, sf ->
sf.channel("messageHandlingExceptionChannel"))
.defaultOutputChannel("exceptionRouterDefaultChannel"));
}
@Bean
public PollableChannel exceptionRouterDefaultChannel() {
return new QueueChannel();
}
@Bean
public PollableChannel illegalArgumentChannel() {
return new QueueChannel();
}
@Bean
public PollableChannel runtimeExceptionChannel() {
return new QueueChannel();
}
@Bean
public PollableChannel messageHandlingExceptionChannel() {
return new QueueChannel();
}
@Bean
public IntegrationFlow routerAsNonLastFlow() {
return f -> f