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

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

Author: artembilan

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

@@ -69,6 +69,7 @@
 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 @@ private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B
 	}
 
 	/**
-	 * 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 B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer) {
 		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 <T> Publisher<Message<T>> toReactivePublisher() {
 		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) {

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

@@ -19,7 +19,10 @@
 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 RouterSpec<K, R> resolutionRequired(boolean resolutionRequired) {
 		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 Collection<Object> getComponentsToRegister() {
 		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 @@ void addMapping(Object key, NamedComponent channel) {
 		}
 
 		@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();
+		public void onApplicationEvent(ContextRefreshedEvent event) {
+			if (event.getApplicationContext() == getApplicationContext() && !this.initialized.getAndSet(true)) {
+				ConversionService conversionService = getConversionService();
+				if (conversionService == null) {
+					conversionService = DefaultConversionService.getSharedInstance();
 				}
-				else if (conversionService.canConvert(key.getClass(), String.class)) {
-					channelKey = conversionService.convert(key, String.class);
+				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());
 				}
-				else {
-					throw new MessagingException("unsupported channel mapping type for router [" + key.getClass() + "]");
-				}
-
-				this.router.setChannelMapping(channelKey, entry.getValue().getComponentName());
 			}
 		}
 

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 @@
 
 /**
  * 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 @@
  */
 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 void setChannelMappings(Map<String, String> channelMappings) {
 	}
 
 	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 @@ private Class<?> resolveClassFromName(String className) {
 	@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 void setChannelMapping(String key, String channelName) {
 	@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 @@ protected List<Object> getChannelKeys(Message<?> message) {
 				cause = cause.getCause();
 			}
 		}
-		return Collections.<Object>singletonList(mostSpecificCause);
+		return Collections.singletonList(mostSpecificCause);
 	}
 
 }

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,23 +50,23 @@
 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
  * @author Gary Russell
  *
  * @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 void testRouterWithTwoSubflows() {
 		@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 void testScatterGather() {
 		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 IntegrationFlow routeSubflowToReplyChannelFlow() {
 			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 IntegrationFlow routeMethodInvocationFlow3() {
 		@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 IntegrationFlow payloadTypeRouteFlow() {
 							.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