INT-4186: Add @Reactive and @ChannelListener

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

Just a PoC how it would look from the end-user perspective

* Add `@ChannelListener` which is fully analogue of similar solution in other places - `@RabbitListener`, `@KafkaListener`, `@JmsListener`.
In other words it has ability to subscriber to several destinations providing common options for target endpoints
The reply will be processed via `@SendTo` or `replyChannel` header.
This functionality is fully similar to the `@StreamListener`
* Add `@Reactive` annotations' option alongside with `@Poller` to allow to use `ReactiveConsumer` underneath with provided backpressure strategy

Author: artembilan

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

@@ -0,0 +1,138 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://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.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+
+/**
+ * Annotation that marks a method to be a listener to inputs channels.
+ *
+ * Annotated methods are allowed to have flexible signatures, which determine how the
+ * method is invoked and how their return results are processed.
+ *
+ * <h3>Individual message handler mode</h3>
+ *
+ * Non declarative methods are treated as message handler based, and are invoked for each
+ * incoming message received from that target. In this case, the method can have a
+ * flexible signature, as described by {@link MessageMapping}.
+ *
+ * If the method returns a {@link org.springframework.messaging.Message}, the result will
+ * be automatically sent to an output channel
+ * (see {@link org.springframework.messaging.handler.annotation.SendTo}), as follows:
+ * <ul>
+ * <li>A result of the type {@link org.springframework.messaging.Message} will be sent
+ * as-is</li>
+ * <li>All other results will become the payload of a
+ * {@link org.springframework.messaging.Message}</li>
+ * </ul>
+ *
+ * The output channel where the return message is sent is determined by consulting
+ * in the following order:
+ * <ul>
+ * <li>The {@link org.springframework.messaging.MessageHeaders} of the resulting
+ * message.</li>
+ * <li>The value set on the
+ * {@link org.springframework.messaging.handler.annotation.SendTo} annotation, if
+ * present</li>
+ * </ul>
+ *
+ * An example of individual message handler signature is as follows:
+ *
+ * <pre>
+ * {@code
+ * @ChannelListener("inputChannel")
+ * @SendTo("outputChannel")
+ * public String convert(String input) {
+ *      return input.toUppercase();
+ * }
+ * }
+ * </pre>
+ *
+ * @author Marius Bogoevici
+ * @author Ilayaperumal Gopinathan
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ * @see org.springframework.messaging.handler.annotation.SendTo
+ */
+@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+@Retention(RetentionPolicy.RUNTIME)
+@MessageMapping
+@Documented
+public @interface ChannelListener {
+
+	/**
+	 * The names of the message channels that the method subscribes to.
+	 */
+	@AliasFor("channel")
+	String[] value() default { };
+
+	/**
+	 * The names of the message channels that the method subscribes to.
+	 */
+	@AliasFor("value")
+	String[] channel() default { };
+
+	/**
+	 * Specify a "chain" of {@code Advice} beans that will "wrap" the message handler.
+	 * Only the handler is advised, not the downstream flow.
+	 * @return the advice chain.
+	 */
+	String[] adviceChain() default { };
+
+	/**
+	 * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option.
+	 * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}.
+	 * Defaults to {@code true}.
+	 * @return the auto startup {@code boolean} flag.
+	 */
+	String autoStartup() default "";
+
+	/**
+	 * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option.
+	 * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer}
+	 * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}.
+	 * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}.
+	 * @return the {@code SmartLifecycle} phase.
+	 */
+	String phase() 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.
+	 * Mutually exclusive with {@link #reactive()}.
+	 */
+	Poller[] poller() default { };
+
+	/**
+	 * @return the {@link Reactive} options for a reactive 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,79 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://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;
+
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.integration.reactive.BackpressureType;
+
+import reactor.core.publisher.BufferOverflowStrategy;
+
+/**
+ * Provides backpressure options for the Messaging annotations for
+ * reactive endpoints.
+ * <p>
+ * It is an analogue of the XML {@code <reactive/>} element.
+ * <p>
+ * Non-reference attributes support Property Placeholder resolutions.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ *
+ * @see reactor.core.publisher.Flux#onBackpressureBuffer
+ * @see reactor.core.publisher.Flux#onBackpressureDrop
+ * @see reactor.core.publisher.Flux#onBackpressureError
+ * @see reactor.core.publisher.Flux#onBackpressureLatest
+ */
+@Target({})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Reactive {
+
+	/**
+	 * @return The {@link BackpressureType} to use.
+	 */
+	@AliasFor("backpressure")
+	BackpressureType value() default BackpressureType.NONE;
+
+	/**
+	 * @return The {@link BackpressureType} to use.
+	 */
+	@AliasFor("value")
+	BackpressureType backpressure() default BackpressureType.NONE;
+
+	/**
+	 * @return The {@link java.util.function.Consumer} bean name, called as a callback on backpressure.
+	 */
+	String consumer() default "";
+
+	/**
+	 * @return The {@link BufferOverflowStrategy} which is used
+	 * in case of {@link BackpressureType#BUFFER} for the {@link #backpressure()}.
+	 */
+	BufferOverflowStrategy bufferOverflowStrategy() default BufferOverflowStrategy.ERROR;
+
+	/**
+	 * @return the maximum buffer backlog size before immediate error
+	 * in case of {@link BackpressureType#BUFFER} for the {@link #backpressure()}.
+	 * Defaults to {@link Integer#MIN_VALUE} meaning {@code unbounded}.
+	 */
+	String bufferMaxSize() default "-2147483648";
+
+}

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

@@ -37,7 +37,7 @@
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
-import org.springframework.integration.channel.MessageChannelReactiveUtils;
+import org.springframework.integration.reactive.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.channel.interceptor.WireTap;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java

@@ -22,7 +22,7 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-import org.springframework.integration.channel.MessageChannelReactiveUtils;
+import org.springframework.integration.reactive.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.Message;
@@ -34,6 +34,7 @@
 import reactor.core.Exceptions;
 import reactor.core.Receiver;
 import reactor.core.Trackable;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Operators;
 
 
@@ -45,7 +46,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
 
 	private final Operators.SubscriberAdapter<Message<?>, Message<?>> subscriber;
 
-	private volatile Publisher<Message<?>> publisher;
+	private volatile Flux<Message<?>> publisher;
 
 	private ErrorHandler errorHandler;
 
@@ -60,7 +61,7 @@ public ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subs
 		Assert.notNull(subscriber);
 
 		Publisher<?> messagePublisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
-		this.publisher = (Publisher<Message<?>>) messagePublisher;
+		this.publisher = Flux.from((Publisher<Message<?>>) messagePublisher);
 
 		this.subscriber = new Operators.SubscriberAdapter<Message<?>, Message<?>>(subscriber) {
 
@@ -93,7 +94,8 @@ protected void onInit() throws Exception {
 
 	@Override
 	protected void doStart() {
-		this.publisher.subscribe(this.subscriber);
+		this.publisher
+				.subscribe(this.subscriber);
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/reactive/BackpressureType.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://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.reactive;
+
+/**
+ * The {@link reactor.core.publisher.Flux} backpressure strategy type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ *
+ * @see org.springframework.integration.annotation.Reactive
+ * @see reactor.core.publisher.Flux#onBackpressureBuffer
+ * @see reactor.core.publisher.Flux#onBackpressureDrop
+ * @see reactor.core.publisher.Flux#onBackpressureError
+ * @see reactor.core.publisher.Flux#onBackpressureLatest
+ */
+public enum BackpressureType {
+
+	BUFFER, DROP, ERROR, LATEST, NONE
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java renamed to spring-integration-core/src/main/java/org/springframework/integration/reactive/MessageChannelReactiveUtils.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel;
+package org.springframework.integration.reactive;
 
 import java.util.Iterator;
 

spring-integration-core/src/main/java/org/springframework/integration/reactive/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides utility classes for Reactive Streams integration.
+ */
+package org.springframework.integration.reactive;

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java

@@ -36,7 +36,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.MessageChannelReactiveUtils;
+import org.springframework.integration.reactive.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.config.EnableIntegration;

spring-integration-core/src/test/java/org/springframework/integration/channellistener/ChannelListenerTests.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://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.channellistener;
+
+import org.junit.runner.RunWith;
+
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.ChannelListener;
+import org.springframework.integration.annotation.Reactive;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.reactive.BackpressureType;
+import org.springframework.stereotype.Service;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import reactor.core.publisher.BufferOverflowStrategy;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class ChannelListenerTests {
+
+	@Configuration
+	@ComponentScan
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+	}
+
+	@Service
+	static class ChannelListenerTestService {
+
+		@ChannelListener(channel = "simpleChannel",
+				reactive = @Reactive(
+						backpressure = BackpressureType.BUFFER,
+						bufferOverflowStrategy = BufferOverflowStrategy.ERROR,
+						bufferMaxSize = "100"))
+		public String simpleListenerTest(String payload) {
+			return payload.toUpperCase();
+		}
+
+	}
+
+}