Add Reactor 11 support

This change adds support for Reactor 1.1 in spring-messaging in
addition to Reactor 1.0.1 -- whichever is present on the classpath is
used.

Note also the module name change:
reactor-tcp:1.0.1 -> reactor-net:1.1.0

Issue: SPR-11636

Author: rstoyanchev

build.gradle

@@ -78,6 +78,7 @@ configure(allprojects) { project ->
 	repositories {
 		maven { url "http://repo.spring.io/libs-release" }
 		maven { url "http://repo.spring.io/milestone" }  // for AspectJ 1.8.0.RC2
+		maven { url "http://repo.spring.io/snapshot" }   // temporarily until Reactor 1.1.0.RC1
 	}
 
 	dependencies {
@@ -396,7 +397,8 @@ project("spring-messaging") {
 		compile(project(":spring-beans"))
 		compile(project(":spring-core"))
 		compile(project(":spring-context"))
-		optional("org.projectreactor:reactor-core:1.0.1.RELEASE")
+		optional("org.projectreactor:reactor-core:1.1.0.BUILD-SNAPSHOT")
+		optional("org.projectreactor:reactor-net:1.1.0.BUILD-SNAPSHOT")
 		optional("org.projectreactor:reactor-tcp:1.0.1.RELEASE")
 		optional("org.eclipse.jetty.websocket:websocket-server:${jettyVersion}") {
 			exclude group: "javax.servlet", module: "javax.servlet-api"
@@ -623,8 +625,9 @@ project("spring-websocket") {
 		testCompile("org.apache.tomcat.embed:tomcat-embed-core:8.0.3")
 		testCompile("org.apache.tomcat.embed:tomcat-embed-websocket:8.0.3")
 		testCompile("org.apache.tomcat.embed:tomcat-embed-logging-juli:8.0.3")
-		testCompile("org.projectreactor:reactor-core:1.0.1.RELEASE")
-		testCompile("org.projectreactor:reactor-tcp:1.0.1.RELEASE")
+
+		testCompile("org.projectreactor:reactor-core:1.1.0.BUILD-SNAPSHOT")
+		testCompile("org.projectreactor:reactor-net:1.1.0.BUILD-SNAPSHOT")
 		testCompile("log4j:log4j:1.2.17")
 		testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}")
 	}

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11StompCodec.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2002-2014 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.messaging.simp.stomp;
+
+import org.springframework.messaging.Message;
+
+import org.springframework.util.Assert;
+import reactor.function.Consumer;
+import reactor.function.Function;
+import reactor.io.Buffer;
+import reactor.io.encoding.Codec;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A Reactor TCP {@link reactor.io.encoding.Codec} for sending and receiving STOMP messages.
+ *
+ * @author Andy Wilkinson
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class Reactor11StompCodec implements Codec<Buffer, Message<byte[]>, Message<byte[]>> {
+
+	private final StompDecoder stompDecoder;
+
+	private final StompEncoder stompEncoder;
+
+	private final Function<Message<byte[]>, Buffer> encodingFunction;
+
+
+	public Reactor11StompCodec() {
+		this(new StompEncoder(), new StompDecoder());
+	}
+
+	public Reactor11StompCodec(StompEncoder encoder, StompDecoder decoder) {
+		Assert.notNull(encoder, "'encoder' is required");
+		Assert.notNull(decoder, "'decoder' is required");
+		this.stompEncoder = encoder;
+		this.stompDecoder = decoder;
+		this.encodingFunction = new EncodingFunction(this.stompEncoder);
+	}
+
+	@Override
+	public Function<Buffer, Message<byte[]>> decoder(final Consumer<Message<byte[]>> messageConsumer) {
+		return new DecodingFunction(this.stompDecoder, messageConsumer);
+	}
+
+	@Override
+	public Function<Message<byte[]>, Buffer> encoder() {
+		return this.encodingFunction;
+	}
+
+
+	private static class EncodingFunction implements Function<Message<byte[]>, Buffer> {
+
+		private final StompEncoder encoder;
+
+		private EncodingFunction(StompEncoder encoder) {
+			this.encoder = encoder;
+		}
+
+		@Override
+		public Buffer apply(Message<byte[]> message) {
+			byte[] bytes = this.encoder.encode(message);
+			return new Buffer(ByteBuffer.wrap(bytes));
+		}
+	}
+
+	private static class DecodingFunction implements Function<Buffer, Message<byte[]>> {
+
+		private final StompDecoder decoder;
+
+		private final Consumer<Message<byte[]>> messageConsumer;
+
+		public DecodingFunction(StompDecoder decoder, Consumer<Message<byte[]>> next) {
+			this.decoder = decoder;
+			this.messageConsumer = next;
+		}
+
+		@Override
+		public Message<byte[]> apply(Buffer buffer) {
+			for (Message<byte[]> message : this.decoder.decode(buffer.byteBuffer())) {
+				this.messageConsumer.accept(message);
+			}
+			return null;
+		}
+	}
+}

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java

@@ -16,7 +16,6 @@
 
 package org.springframework.messaging.simp.stomp;
 
-import java.io.IOException;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.Callable;
@@ -34,8 +33,8 @@
 import org.springframework.messaging.tcp.TcpConnection;
 import org.springframework.messaging.tcp.TcpConnectionHandler;
 import org.springframework.messaging.tcp.TcpOperations;
-import org.springframework.messaging.tcp.reactor.ReactorTcpClient;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
 import org.springframework.util.concurrent.ListenableFutureTask;
@@ -69,6 +68,12 @@
  */
 public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler {
 
+	private static final boolean reactor10Present =
+			ClassUtils.isPresent("reactor.tcp.TcpClient", StompBrokerRelayMessageHandler.class.getClassLoader());
+
+	private static final boolean reactor11Present =
+			ClassUtils.isPresent("reactor.net.tcp.TcpClient", StompBrokerRelayMessageHandler.class.getClassLoader());
+
 	private static final byte[] EMPTY_PAYLOAD = new byte[0];
 
 	private static final ListenableFutureTask<Void> EMPTY_TASK = new ListenableFutureTask<Void>(new VoidCallable());
@@ -331,7 +336,15 @@ protected void startInternal() {
 		this.brokerChannel.subscribe(this);
 
 		if (this.tcpClient == null) {
-			this.tcpClient = new StompTcpClientFactory().create(this.relayHost, this.relayPort);
+			if (reactor11Present) {
+				this.tcpClient = new Reactor11TcpClientFactory().create(this.relayHost, this.relayPort);
+			}
+			else if (reactor10Present) {
+				this.tcpClient = new Reactor10TcpClientFactory().create(this.relayHost, this.relayPort);
+			}
+			else {
+				throw new IllegalStateException("Please add the \"org.projectreactor:reactor-net\" dependency");
+			}
 		}
 
 		if (logger.isDebugEnabled()) {
@@ -606,6 +619,14 @@ public void run() {
 			}
 		}
 
+		@Override
+		public void handleFailure(Throwable ex) {
+			if (this.tcpConnection == null) {
+				return;
+			}
+			handleTcpConnectionFailure("Closing connection after TCP failure", ex);
+		}
+
 		@Override
 		public void afterConnectionClosed() {
 			if (this.tcpConnection == null) {
@@ -753,10 +774,19 @@ public ListenableFuture<Void> forward(Message<?> message) {
 		}
 	}
 
-	private static class StompTcpClientFactory {
+	private static class Reactor11TcpClientFactory {
+
+		public TcpOperations<byte[]> create(String host, int port) {
+			return new org.springframework.messaging.tcp.reactor.Reactor11TcpClient<byte[]>(
+					host, port, new Reactor11StompCodec());
+		}
+	}
+
+	private static class Reactor10TcpClientFactory {
 
-		public TcpOperations<byte[]> create(String relayHost, int relayPort) {
-			return new ReactorTcpClient<byte[]>(relayHost, relayPort, new StompCodec());
+		public TcpOperations<byte[]> create(String host, int port) {
+			return new org.springframework.messaging.tcp.reactor.ReactorTcpClient<byte[]>(
+					host, port, new StompCodec());
 		}
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpConnectionHandler.java

@@ -47,6 +47,12 @@ public interface TcpConnectionHandler<P> {
 	 */
 	void handleMessage(Message<P> message);
 
+	/**
+	 * Handle a failure on the connection.
+	 * @param ex the exception
+	 */
+	void handleFailure(Throwable ex);
+
 	/**
 	 * Invoked after the connection is closed.
 	 */

spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpOperations.java

@@ -50,6 +50,6 @@ public interface TcpOperations<P> {
 	 * @return a ListenableFuture that can be used to determine when and if the
 	 * connection is successfully closed
 	 */
-	ListenableFuture<Void> shutdown();
+	ListenableFuture<Boolean> shutdown();
 
 }

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java

@@ -38,7 +38,7 @@
  * @author Rossen Stoyanchev
  * @since 4.0
  */
-abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements ListenableFuture<T> {
+public abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements ListenableFuture<T> {
 
 	private final Promise<S> promise;
 

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java

@@ -25,7 +25,7 @@
  * @author Rossen Stoyanchev
  * @since 4.0
  */
-class PassThroughPromiseToListenableFutureAdapter<T> extends AbstractPromiseToListenableFutureAdapter<T, T> {
+public class PassThroughPromiseToListenableFutureAdapter<T> extends AbstractPromiseToListenableFutureAdapter<T, T> {
 
 
 	public PassThroughPromiseToListenableFutureAdapter(Promise<T> promise) {