* Rework reactive() attribute of messaging annotations ot a single @Reactive value

with default as `@Reactive(ValueConstants.DEFAULT_NONE)`
* Fix language in docs
* Fix `MessagingAnnotationUtils.resolveAttribute()` to use `requiredType.isInstance()`
instead of comparing classes since annotation instances are `Proxy` at runtime

Author: artembilan

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of aggregating messages.
  * <p>
@@ -109,11 +111,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Messaging Annotation to mark a {@link org.springframework.context.annotation.Bean}
  * method for a {@link org.springframework.messaging.MessageChannel} to produce a
@@ -74,11 +76,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Messaging Annotation to mark a {@link org.springframework.context.annotation.Bean}
  * method for a {@link org.springframework.messaging.MessageChannel} to produce a
@@ -81,11 +83,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of playing the role of a Message Filter.
  * <p>
@@ -132,12 +134,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
-
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -21,7 +21,7 @@
 import java.lang.annotation.Target;
 
 /**
- * Provides a reactive configuration options for the consumer endpoint making
+ * Provides reactive configuration options for the consumer endpoint making
  * any input channel as a reactive stream source of data.
  *
  * @author Artem Bilan

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of resolving to a channel or channel name
  * based on a message, message header(s), or both.
@@ -155,11 +157,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of handling a message or message payload.
  * <p>
@@ -126,11 +128,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of splitting a single message or message
  * payload to produce multiple messages or payloads.
@@ -120,11 +122,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.messaging.handler.annotation.ValueConstants;
+
 /**
  * Indicates that a method is capable of transforming a message, message header,
  * or message payload.
@@ -96,11 +98,10 @@
 	Poller[] poller() default { };
 
 	/**
-	 * @return the {@link Reactive} options for a consumer endpoint.
-	 * This attribute is an {@code array} just to allow an empty default (not reactive).
-	 * Only one {@link Reactive} element is allowed.
+	 * @return the {@link Reactive} marker for a consumer endpoint.
 	 * Mutually exclusive with {@link #poller()}.
+	 * @since 5.5
 	 */
-	Reactive[] reactive() default { };
+	Reactive reactive() default @Reactive(ValueConstants.DEFAULT_NONE);
 
 }

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

@@ -83,6 +83,7 @@
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.core.DestinationResolver;
+import org.springframework.messaging.handler.annotation.ValueConstants;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.support.CronTrigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
@@ -376,41 +377,34 @@ protected AbstractEndpoint createEndpoint(MessageHandler handler, @SuppressWarni
 	protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel,
 			List<Annotation> annotations) {
 
-		AbstractEndpoint endpoint;
-
-
 		Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
-		Reactive[] reactive = MessagingAnnotationUtils.resolveAttribute(annotations, "reactive", Reactive[].class);
+		Reactive reactive = MessagingAnnotationUtils.resolveAttribute(annotations, "reactive", Reactive.class);
+		boolean reactiveProvided = reactive != null && !ValueConstants.DEFAULT_NONE.equals(reactive.value());
 
-		Assert.state(ObjectUtils.isEmpty(reactive) || ObjectUtils.isEmpty(pollers),
+		Assert.state(!reactiveProvided || ObjectUtils.isEmpty(pollers),
 				"The 'poller' and 'reactive' are mutually exclusive.");
 
-		if (inputChannel instanceof Publisher ||
-				handler instanceof ReactiveMessageHandlerAdapter ||
-				!ObjectUtils.isEmpty(reactive)) {
-
-			endpoint = reactiveStreamsConsumer(inputChannel, handler, reactive);
+		if (inputChannel instanceof Publisher || handler instanceof ReactiveMessageHandlerAdapter || reactiveProvided) {
+			return reactiveStreamsConsumer(inputChannel, handler, reactiveProvided ? reactive : null);
 		}
 		else if (inputChannel instanceof SubscribableChannel) {
 			Assert.state(ObjectUtils.isEmpty(pollers), () ->
 					"A '@Poller' should not be specified for Annotation-based " +
 							"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
-			endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
+			return new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
 		}
 		else if (inputChannel instanceof PollableChannel) {
-			endpoint = pollingConsumer(inputChannel, handler, pollers);
+			return pollingConsumer(inputChannel, handler, pollers);
 		}
 		else {
 			throw new IllegalArgumentException("Unsupported 'inputChannel' type: '"
 					+ inputChannel.getClass().getName() + "'. " +
 					"Must be one of 'SubscribableChannel', 'PollableChannel' or 'ReactiveStreamsSubscribableChannel'");
 		}
-
-		return endpoint;
 	}
 
 	private ReactiveStreamsConsumer reactiveStreamsConsumer(MessageChannel channel, MessageHandler handler,
-			Reactive[] reactives) {
+			Reactive reactive) {
 
 		ReactiveStreamsConsumer reactiveStreamsConsumer;
 		if (handler instanceof ReactiveMessageHandlerAdapter) {
@@ -421,10 +415,7 @@ private ReactiveStreamsConsumer reactiveStreamsConsumer(MessageChannel channel,
 			reactiveStreamsConsumer = new ReactiveStreamsConsumer(channel, handler);
 		}
 
-		if (!ObjectUtils.isEmpty(reactives)) {
-			Assert.state(reactives.length == 1,
-					"The 'reactive' for an Annotation-based endpoint can have only one '@Reactive'.");
-			Reactive reactive = reactives[0];
+		if (reactive != null) {
 			String functionBeanName = reactive.value();
 			if (StringUtils.hasText(functionBeanName)) {
 				@SuppressWarnings("unchecked")
@@ -434,7 +425,6 @@ private ReactiveStreamsConsumer reactiveStreamsConsumer(MessageChannel channel,
 			}
 		}
 
-
 		return reactiveStreamsConsumer;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/util/MessagingAnnotationUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -61,11 +61,9 @@ public final class MessagingAnnotationUtils {
 	@SuppressWarnings("unchecked")
 	public static <T> T resolveAttribute(List<Annotation> annotations, String name, Class<T> requiredType) {
 		for (Annotation annotation : annotations) {
-			if (annotation != null) {
-				Object value = AnnotationUtils.getValue(annotation, name);
-				if (value != null && value.getClass() == requiredType && hasValue(value)) {
-					return (T) value;
-				}
+			Object value = AnnotationUtils.getValue(annotation, name);
+			if (requiredType.isInstance(value) && hasValue(value)) {
+				return (T) value;
 			}
 		}
 		return null;

src/reference/asciidoc/dsl.adoc

@@ -265,8 +265,8 @@ The `PollerSpec` is a `FactoryBean` that generates the `PollerMetadata` object f
 [[java-dsl-reactive]]
 === The `reactive()` Endpoint
 
-Starting with version 5.5, the `ConsumerEndpointSpec` provide a `reactive()` configuration property with an optional customizer `Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>>`.
-This option makes the target endpoint as a `ReactiveStreamsConsumer` instance independently of the input channel type, which is turned to the `Flux` via `IntegrationReactiveUtils.messageChannelToFlux()`.
+Starting with version 5.5, the `ConsumerEndpointSpec` provides a `reactive()` configuration property with an optional customizer `Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>>`.
+This option configures the target endpoint as a `ReactiveStreamsConsumer` instance, independently of the input channel type, which is converted to a `Flux` via `IntegrationReactiveUtils.messageChannelToFlux()`.
 The provided function is used from the `Flux.transform()` operator to customize (`publishOn()`, `log()`, `doOnNext()` etc.) a reactive stream source from the input channel.
 
 The following example demonstrates how to change the publishing thread from the input channel independently of the final subscriber and producer to that `DirectChannel`: