Support for reactive result values from event listener methods

Closes gh-21831

spring-context/spring-context.gradle

@@ -2,6 +2,12 @@ description = "Spring Context"
 
 apply plugin: "groovy"
 
+dependencyManagement {
+	imports {
+		mavenBom "io.projectreactor:reactor-bom:${reactorVersion}"
+	}
+}
+
 dependencies {
 	compile(project(":spring-aop"))
 	compile(project(":spring-beans"))
@@ -23,6 +29,8 @@ dependencies {
 	optional("org.hibernate:hibernate-validator:5.4.3.Final")
 	optional("org.jetbrains.kotlin:kotlin-reflect:${kotlinVersion}")
 	optional("org.jetbrains.kotlin:kotlin-stdlib:${kotlinVersion}")
+	optional("org.reactivestreams:reactive-streams")
+	testCompile("io.projectreactor:reactor-core")
 	testCompile("org.codehaus.groovy:groovy-jsr223:${groovyVersion}")
 	testCompile("org.codehaus.groovy:groovy-test:${groovyVersion}")
 	testCompile("org.codehaus.groovy:groovy-xml:${groovyVersion}")

spring-context/src/main/java/org/springframework/context/event/ApplicationListenerMethodAdapter.java

@@ -24,9 +24,12 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CompletionStage;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
 import org.springframework.aop.support.AopUtils;
 import org.springframework.context.ApplicationContext;
@@ -34,20 +37,24 @@ import org.springframework.context.ApplicationEvent;
 import org.springframework.context.PayloadApplicationEvent;
 import org.springframework.context.expression.AnnotatedElementKey;
 import org.springframework.core.BridgeMethodResolver;
+import org.springframework.core.ReactiveAdapter;
+import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.Order;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
+import org.springframework.util.concurrent.ListenableFuture;
 
 /**
  * {@link GenericApplicationListener} adapter that delegates the processing of
  * an event to an {@link EventListener} annotated method.
  *
- * <p>Delegates to {@link #processEvent(ApplicationEvent)} to give sub-classes
+ * <p>Delegates to {@link #processEvent(ApplicationEvent)} to give subclasses
  * a chance to deviate from the default. Unwraps the content of a
  * {@link PayloadApplicationEvent} if necessary to allow a method declaration
  * to define any arbitrary event type. If a condition is defined, it is
@@ -60,6 +67,10 @@ import org.springframework.util.StringUtils;
  */
 public class ApplicationListenerMethodAdapter implements GenericApplicationListener {
 
+	private static final boolean reactiveStreamsPresent = ClassUtils.isPresent(
+			"org.reactivestreams.Publisher", ApplicationListenerMethodAdapter.class.getClassLoader());
+
+
 	protected final Log logger = LogFactory.getLog(getClass());
 
 	private final String beanName;
@@ -213,6 +224,30 @@ public class ApplicationListenerMethodAdapter implements GenericApplicationListe
 	}
 
 	protected void handleResult(Object result) {
+		if (reactiveStreamsPresent && new ReactiveResultHandler().subscribeToPublisher(result)) {
+			if (logger.isTraceEnabled()) {
+				logger.trace("Adapted to reactive result: " + result);
+			}
+		}
+		else if (result instanceof CompletionStage) {
+			((CompletionStage<?>) result).whenComplete((event, ex) -> {
+				if (ex != null) {
+					handleAsyncError(ex);
+				}
+				else if (event != null) {
+					publishEvent(event);
+				}
+			});
+		}
+		else if (result instanceof ListenableFuture) {
+			((ListenableFuture<?>) result).addCallback(this::publishEvents, this::handleAsyncError);
+		}
+		else {
+			publishEvents(result);
+		}
+	}
+
+	private void publishEvents(Object result) {
 		if (result.getClass().isArray()) {
 			Object[] events = ObjectUtils.toObjectArray(result);
 			for (Object event : events) {
@@ -237,6 +272,10 @@ public class ApplicationListenerMethodAdapter implements GenericApplicationListe
 		}
 	}
 
+	protected void handleAsyncError(Throwable t) {
+		logger.error("Unexpected error occurred in asynchronous listener", t);
+	}
+
 	private boolean shouldHandle(ApplicationEvent event, @Nullable Object[] args) {
 		if (args == null) {
 			return false;
@@ -376,4 +415,40 @@ public class ApplicationListenerMethodAdapter implements GenericApplicationListe
 		return this.method.toGenericString();
 	}
 
+
+	private class ReactiveResultHandler {
+
+		public boolean subscribeToPublisher(Object result) {
+			ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(result.getClass());
+			if (adapter != null) {
+				adapter.toPublisher(result).subscribe(new EventPublicationSubscriber());
+				return true;
+			}
+			return false;
+		}
+	}
+
+
+	private class EventPublicationSubscriber implements Subscriber<Object> {
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			s.request(Integer.MAX_VALUE);
+		}
+
+		@Override
+		public void onNext(Object o) {
+			publishEvents(o);
+		}
+
+		@Override
+		public void onError(Throwable t) {
+			handleAsyncError(t);
+		}
+
+		@Override
+		public void onComplete() {
+		}
+	}
+
 }

spring-context/src/main/java/org/springframework/scheduling/support/TaskUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -92,9 +92,7 @@ public abstract class TaskUtils {
 
 		@Override
 		public void handleError(Throwable t) {
-			if (logger.isErrorEnabled()) {
-				logger.error("Unexpected error occurred in scheduled task.", t);
-			}
+			logger.error("Unexpected error occurred in scheduled task", t);
 		}
 	}
 

spring-context/src/test/java/org/springframework/context/event/AnnotationDrivenEventListenerTests.java

@@ -25,6 +25,7 @@ import java.util.Arrays;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.PostConstruct;
@@ -32,6 +33,8 @@ import javax.annotation.PostConstruct;
 import org.junit.After;
 import org.junit.Ignore;
 import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.support.AopUtils;
@@ -61,6 +64,7 @@ import org.springframework.scheduling.annotation.Async;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
+import org.springframework.util.concurrent.SettableListenableFuture;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.validation.beanvalidation.MethodValidationPostProcessor;
 
@@ -243,7 +247,69 @@ public class AnnotationDrivenEventListenerTests {
 	}
 
 	@Test
-	public void eventListenerWorksWithSimpleInterfaceProxy() throws Exception {
+	public void listenableFutureReply() {
+		load(TestEventListener.class, ReplyEventListener.class);
+		SettableListenableFuture<String> future = new SettableListenableFuture<>();
+		future.set("dummy");
+		AnotherTestEvent event = new AnotherTestEvent(this, future);
+		ReplyEventListener replyEventListener = this.context.getBean(ReplyEventListener.class);
+		TestEventListener listener = this.context.getBean(TestEventListener.class);
+
+		this.eventCollector.assertNoEventReceived(listener);
+		this.eventCollector.assertNoEventReceived(replyEventListener);
+		this.context.publishEvent(event);
+		this.eventCollector.assertEvent(replyEventListener, event);
+		this.eventCollector.assertEvent(listener, "dummy"); // reply
+		this.eventCollector.assertTotalEventsCount(2);
+	}
+
+	@Test
+	public void completableFutureReply() {
+		load(TestEventListener.class, ReplyEventListener.class);
+		AnotherTestEvent event = new AnotherTestEvent(this, CompletableFuture.completedFuture("dummy"));
+		ReplyEventListener replyEventListener = this.context.getBean(ReplyEventListener.class);
+		TestEventListener listener = this.context.getBean(TestEventListener.class);
+
+		this.eventCollector.assertNoEventReceived(listener);
+		this.eventCollector.assertNoEventReceived(replyEventListener);
+		this.context.publishEvent(event);
+		this.eventCollector.assertEvent(replyEventListener, event);
+		this.eventCollector.assertEvent(listener, "dummy"); // reply
+		this.eventCollector.assertTotalEventsCount(2);
+	}
+
+	@Test
+	public void monoReply() {
+		load(TestEventListener.class, ReplyEventListener.class);
+		AnotherTestEvent event = new AnotherTestEvent(this, Mono.just("dummy"));
+		ReplyEventListener replyEventListener = this.context.getBean(ReplyEventListener.class);
+		TestEventListener listener = this.context.getBean(TestEventListener.class);
+
+		this.eventCollector.assertNoEventReceived(listener);
+		this.eventCollector.assertNoEventReceived(replyEventListener);
+		this.context.publishEvent(event);
+		this.eventCollector.assertEvent(replyEventListener, event);
+		this.eventCollector.assertEvent(listener, "dummy"); // reply
+		this.eventCollector.assertTotalEventsCount(2);
+	}
+
+	@Test
+	public void fluxReply() {
+		load(TestEventListener.class, ReplyEventListener.class);
+		AnotherTestEvent event = new AnotherTestEvent(this, Flux.just("dummy1", "dummy2"));
+		ReplyEventListener replyEventListener = this.context.getBean(ReplyEventListener.class);
+		TestEventListener listener = this.context.getBean(TestEventListener.class);
+
+		this.eventCollector.assertNoEventReceived(listener);
+		this.eventCollector.assertNoEventReceived(replyEventListener);
+		this.context.publishEvent(event);
+		this.eventCollector.assertEvent(replyEventListener, event);
+		this.eventCollector.assertEvent(listener, "dummy1", "dummy2"); // reply
+		this.eventCollector.assertTotalEventsCount(3);
+	}
+
+	@Test
+	public void eventListenerWorksWithSimpleInterfaceProxy() {
 		load(ScopedProxyTestBean.class);
 
 		SimpleService proxy = this.context.getBean(SimpleService.class);
@@ -260,7 +326,7 @@ public class AnnotationDrivenEventListenerTests {
 	}
 
 	@Test
-	public void eventListenerWorksWithAnnotatedInterfaceProxy() throws Exception {
+	public void eventListenerWorksWithAnnotatedInterfaceProxy() {
 		load(AnnotatedProxyTestBean.class);
 
 		AnnotatedSimpleService proxy = this.context.getBean(AnnotatedSimpleService.class);
@@ -277,7 +343,7 @@ public class AnnotationDrivenEventListenerTests {
 	}
 
 	@Test
-	public void eventListenerWorksWithCglibProxy() throws Exception {
+	public void eventListenerWorksWithCglibProxy() {
 		load(CglibProxyTestBean.class);
 
 		CglibProxyTestBean proxy = this.context.getBean(CglibProxyTestBean.class);
@@ -294,14 +360,14 @@ public class AnnotationDrivenEventListenerTests {
 	}
 
 	@Test
-	public void privateMethodOnCglibProxyFails() throws Exception {
+	public void privateMethodOnCglibProxyFails() {
 		assertThatExceptionOfType(BeanInitializationException.class).isThrownBy(() ->
 				load(CglibProxyWithPrivateMethod.class))
 			.withCauseInstanceOf(IllegalStateException.class);
 	}
 
 	@Test
-	public void eventListenerWorksWithCustomScope() throws Exception {
+	public void eventListenerWorksWithCustomScope() {
 		load(CustomScopeTestBean.class);
 		CustomScope customScope = new CustomScope();
 		this.context.getBeanFactory().registerScope("custom", customScope);

spring-context/src/test/java/org/springframework/context/event/ApplicationListenerMethodAdapterTests.java

@@ -99,8 +99,7 @@ public class ApplicationListenerMethodAdapterTests extends AbstractApplicationEv
 
 	@Test
 	public void listenerWithSubTypeSeveralGenerics() {
-		Method method = ReflectionUtils.findMethod(SampleEvents.class,
-				"handleString", String.class);
+		Method method = ReflectionUtils.findMethod(SampleEvents.class, "handleString", String.class);
 		supportsEventType(true, method, ResolvableType.forClass(PayloadTestEvent.class));
 	}
 
@@ -141,23 +140,20 @@ public class ApplicationListenerMethodAdapterTests extends AbstractApplicationEv
 	public void listenerWithTooManyParameters() {
 		Method method = ReflectionUtils.findMethod(
 				SampleEvents.class, "tooManyParameters", String.class, String.class);
-		assertThatIllegalStateException().isThrownBy(() ->
-				createTestInstance(method));
+		assertThatIllegalStateException().isThrownBy(() -> createTestInstance(method));
 	}
 
 	@Test
 	public void listenerWithNoParameter() {
 		Method method = ReflectionUtils.findMethod(SampleEvents.class, "noParameter");
-		assertThatIllegalStateException().isThrownBy(() ->
-				createTestInstance(method));
+		assertThatIllegalStateException().isThrownBy(() -> createTestInstance(method));
 	}
 
 	@Test
 	public void listenerWithMoreThanOneParameter() {
 		Method method = ReflectionUtils.findMethod(
 				SampleEvents.class, "moreThanOneParameter", String.class, Integer.class);
-		assertThatIllegalStateException().isThrownBy(() ->
-				createTestInstance(method));
+		assertThatIllegalStateException().isThrownBy(() -> createTestInstance(method));
 	}
 
 	@Test
@@ -331,7 +327,8 @@ public class ApplicationListenerMethodAdapterTests extends AbstractApplicationEv
 
 	private void supportsEventType(boolean match, Method method, ResolvableType eventType) {
 		ApplicationListenerMethodAdapter adapter = createTestInstance(method);
-		assertThat(adapter.supportsEventType(eventType)).as("Wrong match for event '" + eventType + "' on " + method).isEqualTo(match);
+		assertThat(adapter.supportsEventType(eventType))
+				.as("Wrong match for event '" + eventType + "' on " + method).isEqualTo(match);
 	}
 
 	private void invokeListener(Method method, ApplicationEvent event) {

spring-context/src/test/java/org/springframework/context/event/EventPublicationInterceptorTests.java

@@ -16,7 +16,6 @@
 
 package org.springframework.context.event;
 
-import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.aop.framework.ProxyFactory;
@@ -42,16 +41,11 @@ import static org.mockito.Mockito.mock;
  */
 public class EventPublicationInterceptorTests {
 
-	private ApplicationEventPublisher publisher;
+	private final ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
 
 
-	@Before
-	public void setUp() {
-		this.publisher = mock(ApplicationEventPublisher.class);
-	}
-
 	@Test
-	public void testWithNoApplicationEventClassSupplied() throws Exception {
+	public void testWithNoApplicationEventClassSupplied() {
 		EventPublicationInterceptor interceptor = new EventPublicationInterceptor();
 		interceptor.setApplicationEventPublisher(this.publisher);
 		assertThatIllegalArgumentException().isThrownBy(
@@ -59,7 +53,7 @@ public class EventPublicationInterceptorTests {
 	}
 
 	@Test
-	public void testWithNonApplicationEventClassSupplied() throws Exception {
+	public void testWithNonApplicationEventClassSupplied() {
 		EventPublicationInterceptor interceptor = new EventPublicationInterceptor();
 		interceptor.setApplicationEventPublisher(this.publisher);
 		assertThatIllegalArgumentException().isThrownBy(() -> {
@@ -69,7 +63,7 @@ public class EventPublicationInterceptorTests {
 	}
 
 	@Test
-	public void testWithAbstractStraightApplicationEventClassSupplied() throws Exception {
+	public void testWithAbstractStraightApplicationEventClassSupplied() {
 		EventPublicationInterceptor interceptor = new EventPublicationInterceptor();
 		interceptor.setApplicationEventPublisher(this.publisher);
 		assertThatIllegalArgumentException().isThrownBy(() -> {
@@ -79,7 +73,7 @@ public class EventPublicationInterceptorTests {
 	}
 
 	@Test
-	public void testWithApplicationEventClassThatDoesntExposeAValidCtor() throws Exception {
+	public void testWithApplicationEventClassThatDoesntExposeAValidCtor() {
 		EventPublicationInterceptor interceptor = new EventPublicationInterceptor();
 		interceptor.setApplicationEventPublisher(this.publisher);
 		assertThatIllegalArgumentException().isThrownBy(() -> {
@@ -89,7 +83,7 @@ public class EventPublicationInterceptorTests {
 	}
 
 	@Test
-	public void testExpectedBehavior() throws Exception {
+	public void testExpectedBehavior() {
 		TestBean target = new TestBean();
 		final TestListener listener = new TestListener();