INT-3945: Add async to the @serviceactivator

JIRA: https://jira.spring.io/browse/INT-3945

Adding test for async annotated ServiceActivator

fix whitespace

Address comments

Change copyright and author

* Some code style polishing
* Rename `Log4j2LevelAdjuster.level()` to more friendly `forLevel()`
factory method name.
* Add `@param level` to the `Log4j2LevelAdjuster.forLevel()` to fix
JavaDoc warning

Author: yilinwei

spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -41,8 +41,9 @@
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Yilin Wei
  */
-@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
+@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
 @Retention(RetentionPolicy.RUNTIME)
 @Inherited
 @Documented
@@ -77,7 +78,7 @@
 	 * Only the handler is advised, not the downstream flow.
 	 * @return the advice chain.
 	 */
-	String[] adviceChain() default {};
+	String[] adviceChain() default { };
 
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
@@ -110,12 +111,19 @@
 	 */
 	String phase() default "";
 
+	/**
+	 * Specify whether the service method is async.
+	 * This value is {@code false} by default.
+	 * @return the async flag.
+	 */
+	String async() default "";
+
 	/**
 	 * @return the {@link Poller} options for a polled endpoint
 	 * ({@link org.springframework.integration.scheduling.PollerMetadata}).
 	 * This attribute is an {@code array} just to allow an empty default (no poller).
 	 * Only one {@link Poller} element is allowed.
 	 */
-	Poller[] poller() default {};
+	Poller[] poller() default { };
 
 }

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -257,7 +257,7 @@ protected List<Advice> extractAdviceChain(String beanName, List<Annotation> anno
 		 * by setting an empty array on the custom annotation.
 		 */
 		if (adviceChainNames != null && adviceChainNames.length > 0) {
-			adviceChain = new ArrayList<Advice>();
+			adviceChain = new ArrayList<>();
 			for (String adviceChainName : adviceChainNames) {
 				Object adviceChainBean = this.beanFactory.getBean(adviceChainName);
 				if (adviceChainBean instanceof Advice) {

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -38,12 +38,13 @@
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Yilin Wei
  */
 public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<ServiceActivator> {
 
 	public ServiceActivatorAnnotationPostProcessor(ConfigurableListableBeanFactory beanFactory) {
 		super(beanFactory);
-		this.messageHandlerAttributes.addAll(Arrays.<String>asList("outputChannel", "requiresReply", "adviceChain"));
+		this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "requiresReply", "adviceChain"));
 	}
 
 
@@ -56,9 +57,9 @@ protected MessageHandler createHandler(Object bean, Method method, List<Annotati
 			if (serviceActivator == null) {
 				if (target instanceof MessageHandler) {
 					/*
-			 		 * Return a reply-producing message handler so that we still get 'produced no reply' messages
-			 		 * and the super class will inject the advice chain to advise the handler method if needed.
-			 		 */
+					 * Return a reply-producing message handler so that we still get 'produced no reply' messages
+					 * and the super class will inject the advice chain to advise the handler method if needed.
+					 */
 					return new ReplyProducingMessageHandlerWrapper((MessageHandler) target);
 				}
 				else {
@@ -79,6 +80,11 @@ protected MessageHandler createHandler(Object bean, Method method, List<Annotati
 			serviceActivator.setRequiresReply(Boolean.parseBoolean(this.beanFactory.resolveEmbeddedValue(requiresReply)));
 		}
 
+		String isAsync = MessagingAnnotationUtils.resolveAttribute(annotations, "async", String.class);
+		if (StringUtils.hasText(isAsync)) {
+			serviceActivator.setAsync(Boolean.parseBoolean(this.beanFactory.resolveEmbeddedValue(isAsync)));
+		}
+
 		this.setOutputChannelIfPresent(annotations, serviceActivator);
 		return serviceActivator;
 	}

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml

@@ -19,10 +19,20 @@
 	<beans:bean id="annotatedEndpoint2"
 			class="org.springframework.integration.config.annotation.AnnotatedEndpointActivationTests.AnnotatedEndpoint2"/>
 
+	<beans:bean id="annotatedEndpoint3"
+		        class="org.springframework.integration.config.annotation.AnnotatedEndpointActivationTests.AnnotatedEndpoint3"/>
+
 	<channel id="input"/>
 
 	<channel id="output">
 		<queue capacity="10"/>
 	</channel>
 
+
+	<channel id="inputAsync"/>
+
+	<channel id="outputAsync">
+		<queue capacity="10"/>
+	</channel>
+
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -35,17 +35,18 @@
 import org.springframework.messaging.PollableChannel;
 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;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Dave Syer
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Yilin Wei
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class AnnotatedEndpointActivationTests {
 
@@ -57,6 +58,14 @@ public class AnnotatedEndpointActivationTests {
 	@Qualifier("output")
 	private PollableChannel output;
 
+	@Autowired
+	@Qualifier("inputAsync")
+	private MessageChannel inputAsync;
+
+	@Autowired
+	@Qualifier("outputAsync")
+	private PollableChannel outputAsync;
+
 	@Autowired
 	private AbstractApplicationContext applicationContext;
 
@@ -72,7 +81,7 @@ public void resetCount() {
 
 	@Test
 	public void sendAndReceive() {
-		this.input.send(new GenericMessage<String>("foo"));
+		this.input.send(new GenericMessage<>("foo"));
 		Message<?> message = this.output.receive(100);
 		assertNotNull(message);
 		assertEquals("foo: 1", message.getPayload());
@@ -82,10 +91,19 @@ public void sendAndReceive() {
 		assertTrue(this.applicationContext.containsBean("annotatedEndpoint2.process.serviceActivator"));
 	}
 
+	@Test
+	public void sendAndReceiveAsync() {
+		this.inputAsync.send(new GenericMessage<>("foo"));
+		Message<?> message = this.outputAsync.receive(100);
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+		assertTrue(this.applicationContext.containsBean("annotatedEndpoint3.process.serviceActivator"));
+	}
+
 	@Test
 	public void sendAndReceiveImplicitInputChannel() {
 		MessageChannel input = this.applicationContext.getBean("inputImplicit", MessageChannel.class);
-		input.send(new GenericMessage<String>("foo"));
+		input.send(new GenericMessage<>("foo"));
 		Message<?> message = this.output.receive(100);
 		assertNotNull(message);
 		assertEquals("foo: 1", message.getPayload());
@@ -96,15 +114,15 @@ public void sendAndReceiveImplicitInputChannel() {
 	@DirtiesContext
 	public void stopContext() {
 		applicationContext.stop();
-		this.input.send(new GenericMessage<String>("foo"));
+		this.input.send(new GenericMessage<>("foo"));
 	}
 
 	@Test
 	@DirtiesContext
 	public void stopAndRestartContext() {
 		applicationContext.stop();
 		applicationContext.start();
-		this.input.send(new GenericMessage<String>("foo"));
+		this.input.send(new GenericMessage<>("foo"));
 		Message<?> message = this.output.receive(100);
 		assertNotNull(message);
 		assertEquals("foo: 1", message.getPayload());
@@ -139,5 +157,16 @@ public String process(String message) {
 
 	}
 
+	@SuppressWarnings("unused")
+	private static class AnnotatedEndpoint3 {
+
+		@ServiceActivator(inputChannel = "inputAsync", outputChannel = "outputAsync", async = "true")
+		public ListenableFuture<String> process(String message) {
+			SettableListenableFuture<String> future = new SettableListenableFuture<>();
+			future.set(message);
+			return future;
+		}
+
+	}
 
 }

spring-integration-test-support/src/main/java/org/springframework/integration/test/rule/Log4j2LevelAdjuster.java

@@ -183,7 +183,7 @@ public Log4j2LevelAdjuster categories(String... categories) {
 	 * @return the Log4j2LevelAdjuster instance
 	 */
 	public static Log4j2LevelAdjuster trace() {
-		return level(Level.TRACE);
+		return forLevel(Level.TRACE);
 	}
 
 	/**
@@ -192,7 +192,7 @@ public static Log4j2LevelAdjuster trace() {
 	 * @return the Log4j2LevelAdjuster instance
 	 */
 	public static Log4j2LevelAdjuster debug() {
-		return level(Level.DEBUG);
+		return forLevel(Level.DEBUG);
 	}
 
 	/**
@@ -201,15 +201,16 @@ public static Log4j2LevelAdjuster debug() {
 	 * @return the Log4j2LevelAdjuster instance
 	 */
 	public static Log4j2LevelAdjuster info() {
-		return level(Level.INFO);
+		return forLevel(Level.INFO);
 	}
 
 	/**
 	 * The factory to produce Log4j2LevelAdjuster instances for arbitrary logging {@link Level}
 	 * with the {@code org.springframework.integration} as default category.
+	 * @param level the {@link Level} to use for logging
 	 * @return the Log4j2LevelAdjuster instance
 	 */
-	public static Log4j2LevelAdjuster level(Level level) {
+	public static Log4j2LevelAdjuster forLevel(Level level) {
 		return new Log4j2LevelAdjuster(level);
 	}