Add initial support for RSockets (#2902)

* Add initial support for RSockets

* Add `spring-integration-rsocket` module and respective dependencies
* Implement `RSocketOutboundGateway` based on the Spring Messaging
`RSocketRequester`.
This component supports dynamic RSocket properties via expressions
against request message.
to handle `Publisher` for requests, it must be present in the request
message `payload` instead of `FluxMessageChannel` upstream, since the
last one just flattens events to be handled in the `MessageHandler` one
by one.
The result `Mono` is subscribed downstream in the `FluxMessageChannel`
or directly by the `AbstractReplyProducingMessageHandler`.
If result is a `Flux` it is just wrapped into the `Mono` to be processed
downstream by end-user code.
The point is that these request/replies are volatile and live in the
particular context meanwhile a `FluxMessageChannel` is long living
publisher in the application context boundaries.
* The `RSocketOutboundGatewayIntegrationTests` is an adapted copy of
`RSocketClientToServerIntegrationTests` from Spring Messaging
* Add `doOnError()` into the `Flux` created in the
`AbstractMessageProducingHandler` for `Publisher` replies

* * Use singular for the `RSocket` term
* Use no-op `Consumer` for the `strategiesConfigurer` and
`factoryConfigurer` in the `RSocketOutboundGateway` and also
`Assert.notNull()` in the appropriate setters to avoid null check during
`RSocketRequester.builder()` initialization
* Use `TcpServer.create().port(0)` in the
`RSocketOutboundGatewayIntegrationTests` to allow to select free OS port
and bind into it.
The selected port is used later for client configuration in the
`RSocketOutboundGateway` bean definition

* * Change `RSocketOutboundGatewayIntegrationTests.PORT` to lower case

build.gradle

@@ -39,6 +39,7 @@ allprojects {
 		if (version.endsWith('BUILD-SNAPSHOT')) {
 			maven { url 'https://repo.spring.io/libs-snapshot' }
 		}
+        maven { url "https://oss.jfrog.org/artifactory/libs-snapshot" } // RSocket
 //		maven { url 'https://repo.spring.io/libs-staging-local' }
 	}
 
@@ -128,10 +129,11 @@ subprojects { subproject ->
 		mysqlVersion = '8.0.15'
 		pahoMqttClientVersion = '1.2.0'
 		postgresVersion = '42.2.5'
-		reactorNettyVersion = '0.8.6.RELEASE'
-		reactorVersion = '3.2.8.RELEASE'
+		reactorNettyVersion = '0.9.0.BUILD-SNAPSHOT'
+		reactorVersion = '3.3.0.BUILD-SNAPSHOT'
 		resilience4jVersion = '0.14.1'
 		romeToolsVersion = '1.12.0'
+		rsocketVersion = '0.12.2-RC3-SNAPSHOT'
 		servletApiVersion = '4.0.1'
 		smackVersion = '4.3.3'
 		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.2.0.M1'
@@ -141,7 +143,7 @@ subprojects { subproject ->
 		springGemfireVersion = '2.2.0.M3'
 		springSecurityVersion = '5.2.0.M2'
 		springRetryVersion = '1.2.4.RELEASE'
-		springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.0.M1'
+		springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.0.BUILD-SNAPSHOT'
 		springWsVersion = '3.0.7.RELEASE'
 		tomcatVersion = "9.0.17"
 		xstreamVersion = '1.4.11.1'
@@ -596,6 +598,18 @@ project('spring-integration-rmi') {
 	}
 }
 
+project('spring-integration-rsocket') {
+	description = 'Spring Integration RSocket Support'
+	dependencies {
+		compile project(":spring-integration-core")
+		compile("io.projectreactor.netty:reactor-netty:$reactorNettyVersion")
+		compile("io.rsocket:rsocket-core:$rsocketVersion")
+		compile("io.rsocket:rsocket-transport-netty:$rsocketVersion")
+
+		testCompile "io.projectreactor:reactor-test:$reactorVersion"
+	}
+}
+
 project('spring-integration-scripting') {
 	description = 'Spring Integration Scripting Support'
 	dependencies {

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -82,8 +82,8 @@ public void subscribeTo(Publisher<Message<?>> publisher) {
 		ConnectableFlux<?> connectableFlux =
 				Flux.from(publisher)
 						.handle((message, sink) -> sink.next(send(message)))
-						.onErrorContinue((throwable, o) -> logger.warn("Error during processing event: " + o, throwable)
-						)
+						.onErrorContinue((throwable, event) ->
+								logger.warn("Error during processing event: " + event, throwable))
 						.doOnComplete(() -> this.publishers.remove(publisher))
 						.publish();
 

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -29,6 +29,7 @@
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.routingslip.RoutingSlipRouteStrategy;
@@ -283,6 +284,7 @@ private void doProduceOutput(final Message<?> requestMessage, final MessageHeade
 				((ReactiveStreamsSubscribableChannel) messageChannel)
 						.subscribeTo(
 								Flux.from((Publisher<?>) reply)
+										.doOnError((ex) -> sendErrorMessage(requestMessage, ex))
 										.map(result -> createOutputMessage(result, requestHeaders)));
 			}
 		}
@@ -311,25 +313,22 @@ else if (reply instanceof AbstractIntegrationMessageBuilder) {
 	}
 
 	private void asyncNonReactiveReply(Message<?> requestMessage, Object reply, Object replyChannel) {
-
 		ListenableFuture<?> future;
 		if (reply instanceof ListenableFuture<?>) {
 			future = (ListenableFuture<?>) reply;
 		}
 		else {
 			SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
-
 			Mono.from((Publisher<?>) reply)
 					.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
-
 			future = settableListenableFuture;
 		}
-
 		future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel));
 	}
 
 	private Object getOutputChannelFromRoutingSlip(Object reply, Message<?> requestMessage, List<?> routingSlip,
 			AtomicInteger routingSlipIndex) {
+
 		if (routingSlipIndex.get() >= routingSlip.size()) {
 			return null;
 		}
@@ -365,7 +364,7 @@ else if (path instanceof RoutingSlipRouteStrategy) {
 	}
 
 	protected Message<?> createOutputMessage(Object output, MessageHeaders requestHeaders) {
-		AbstractIntegrationMessageBuilder<?> builder = null;
+		AbstractIntegrationMessageBuilder<?> builder;
 		if (output instanceof Message<?>) {
 			if (this.noHeadersPropagation || !shouldCopyRequestHeaders()) {
 				return (Message<?>) output;
@@ -449,7 +448,7 @@ protected void sendErrorMessage(Message<?> requestMessage, Throwable ex) {
 			}
 			catch (Exception e) {
 				Exception exceptionToLog =
-						IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, () -> null,  e);
+						IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, () -> null, e);
 				logger.error("Failed to send async reply", exceptionToLog);
 			}
 		}
@@ -459,7 +458,7 @@ protected Object resolveErrorChannel(final MessageHeaders requestHeaders) {
 		Object errorChannel = requestHeaders.getErrorChannel();
 		if (errorChannel == null) {
 			try {
-				errorChannel = getChannelResolver().resolveDestination("errorChannel");
+				errorChannel = getChannelResolver().resolveDestination(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
 			}
 			catch (DestinationResolutionException e) {
 				// ignore

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java

@@ -87,6 +87,9 @@ public void setBeanClassLoader(ClassLoader beanClassLoader) {
 		this.beanClassLoader = beanClassLoader;
 	}
 
+	protected ClassLoader getBeanClassLoader() {
+		return this.beanClassLoader;
+	}
 
 	@Override
 	protected final void onInit() {

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 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.
+ * 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.rsocket.config;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * Namespace handler for Spring Integration's <em>RSocket</em> namespace.
+ *
+ * @author Artem Bilan
+ */
+public class RSocketNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	public void init() {
+
+	}
+
+}