diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 69a5ca7598..e958181bca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -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 * {@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 + * {@code + * .routeByException(r -> r + * .channelMapping(IllegalArgumentException.class, "illegalArgumentChannel") + * .subFlowMapping(MessageHandlingException.class, sf -> + * sf.handle(...)) + * ) + * } + * + * @param routerConfigurer the {@link Consumer} to provide {@link ErrorMessageExceptionTypeRouter} options. + * @return the current {@link IntegrationFlowDefinition}. + * @see ErrorMessageExceptionTypeRouter + */ + public B routeByException( + Consumer, 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(this.integrationComponents, publisher); } + @SuppressWarnings("unchecked") private > B register(S endpointSpec, Consumer endpointConfigurer) { if (endpointConfigurer != null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java index ebec8af1cb..e3eb909053 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java @@ -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 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. + *

NOTE: this does not affect routing, just the reporting which dynamically + * resolved channels have been routed to. Default {@code 100}. + * @param dynamicChannelLimit the limit. + * @return the router spec. + * @see AbstractMappingMessageRouter#setDynamicChannelLimit(int) + */ + public RouterSpec 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 return super.getComponentsToRegister(); } - private static class RouterMappingProvider extends IntegrationObjectSupport { + private static class RouterMappingProvider extends IntegrationObjectSupport + implements ApplicationListener { + + private final AtomicBoolean initialized = new AtomicBoolean(); private final MappingMessageRouterManagement router; @@ -178,28 +198,31 @@ public final class RouterSpec } @Override - protected void onInit() throws Exception { - ConversionService conversionService = getConversionService(); - if (conversionService == null) { - conversionService = DefaultConversionService.getSharedInstance(); - } - for (Map.Entry 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 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()); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java index d2e644b89a..18a5de570a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java @@ -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. *

* 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> classNameMappings = new ConcurrentHashMap>(); + private volatile Map> classNameMappings = new ConcurrentHashMap<>(); private volatile boolean initialized; @@ -57,7 +58,7 @@ public class ErrorMessageExceptionTypeRouter extends AbstractMappingMessageRoute } private void populateClassNameMapping(Set classNames) { - Map> newClassNameMappings = new ConcurrentHashMap>(); + Map> 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> newClassNameMappings = new ConcurrentHashMap>(this.classNameMappings); + Map> 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> newClassNameMappings = new ConcurrentHashMap>(this.classNameMappings); + Map> 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.singletonList(mostSpecificCause); + return Collections.singletonList(mostSpecificCause); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java index 6d2d7f1fec..27d60926ac 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java @@ -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 results = (List) 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 .route("true", m -> m .subFlowMapping(true, sf -> sf - .handle((p, h) -> p.toUpperCase()) - ) + .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