INT-4444: Introduce @Reactive & reactive()

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

Right now the high-level API creates a `ReactiveStreamsConsumer`
only when the input channel is a `Publisher<?>` impl or target handler
is a `ReactiveMessageHandler`

* Add `@Reactive[] reactive()` attribute to messaging annotations
* Add `ConsumerEndpointSpec.reactive()`
Both options point to the same `ConsumerEndpointFactoryBean.setReactiveCustomizer()`
making the target endpoint always as a `ReactiveStreamsConsumer` independently of
the input channel and target handler
* Use the `Function` to customize a source `Flux` from the channel
* Test and document a new feature

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -104,8 +104,16 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeFrom.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.
@@ -69,8 +69,16 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeTo.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.
@@ -50,8 +50,8 @@
 public @interface BridgeTo {
 
 	/**
-	 * @return the outbound channel name to send the message to
-	 * {@link org.springframework.integration.handler.BridgeHandler}.
+	 * @return the outbound channel name to send the message to for the
+	 * {@link org.springframework.integration.handler.BridgeHandler} reply.
 	 * Optional: when omitted the message is sent to the {@code reply-channel}
 	 * in its headers (if present - an exception is thrown otherwise).
 	 */
@@ -73,11 +73,19 @@
 	String phase() default "";
 
 	/**
-	 * @return the {@link org.springframework.integration.annotation.Poller} options for a polled endpoint
+	 * @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 org.springframework.integration.annotation.Poller} element is allowed.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -38,6 +38,7 @@
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 @Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@@ -126,8 +127,17 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
+
 }

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.annotation;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Provides a reactive configuration options for the consumer endpoint making
+ * any input channel as a reactive stream source of data.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.5
+ */
+@Target({})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Reactive {
+
+	/**
+	 * @return the function bean name to be used in the
+	 * {@link reactor.core.publisher.Flux#transform} on the input channel
+	 * {@link reactor.core.publisher.Flux}.
+	 */
+	String value() default "";
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -150,8 +150,16 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -121,7 +121,16 @@
 	 * ({@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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -115,8 +115,16 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -91,8 +91,16 @@
 	 * @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.
+	 * Mutually exclusive with {@link #reactive()}.
 	 */
 	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.
+	 * Mutually exclusive with {@link #poller()}.
+	 */
+	Reactive[] reactive() default { };
+
 }