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

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.FixedIntervalReconnectStrategy;
 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 @@ import org.springframework.util.concurrent.ListenableFutureTask;
  */
 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 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 		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 class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 			}
 		}
 
+		@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 class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 		}
 	}
 
-	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 @@ import reactor.function.Consumer;
  * @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 @@ import reactor.core.composable.Promise;
  * @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) {

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

@@ -0,0 +1,218 @@
+/*
+ * 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.tcp.reactor;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Properties;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.tcp.ReconnectStrategy;
+import org.springframework.messaging.tcp.TcpConnectionHandler;
+import org.springframework.messaging.tcp.TcpOperations;
+import org.springframework.util.Assert;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import reactor.core.Environment;
+import reactor.core.composable.Composable;
+import reactor.core.composable.Promise;
+import reactor.core.composable.Stream;
+import reactor.core.composable.spec.Promises;
+import reactor.core.configuration.ConfigurationReader;
+import reactor.core.configuration.DispatcherConfiguration;
+import reactor.core.configuration.ReactorConfiguration;
+import reactor.function.Consumer;
+import reactor.function.Function;
+import reactor.io.Buffer;
+import reactor.io.encoding.Codec;
+import reactor.net.NetChannel;
+import reactor.net.Reconnect;
+import reactor.net.netty.tcp.NettyTcpClient;
+import reactor.net.tcp.TcpClient;
+import reactor.net.tcp.spec.TcpClientSpec;
+import reactor.tuple.Tuple;
+import reactor.tuple.Tuple2;
+
+
+/**
+ * An implementation of {@link org.springframework.messaging.tcp.TcpOperations}
+ * based on the TCP client support of the Reactor project.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class Reactor11TcpClient<P> implements TcpOperations<P> {
+
+	public static final Class<NettyTcpClient> REACTOR_TCP_CLIENT_TYPE = NettyTcpClient.class;
+
+
+	private final TcpClient<Message<P>, Message<P>> tcpClient;
+
+	private final Environment environment;
+
+
+	/**
+	 * A constructor that creates a {@link reactor.net.netty.tcp.NettyTcpClient} with
+	 * a {@link reactor.event.dispatch.SynchronousDispatcher} as a result of which
+	 * network I/O is handled in Netty threads.
+	 *
+	 * <p>Also see the constructor accepting a pre-configured Reactor
+	 * {@link reactor.net.tcp.TcpClient}.
+	 *
+	 * @param host the host to connect to
+	 * @param port the port to connect to
+	 * @param codec the codec to use for encoding and decoding the TCP stream
+	 */
+	public Reactor11TcpClient(String host, int port, Codec<Buffer, Message<P>, Message<P>> codec) {
+
+		// Revisit in 1.1: is Environment still required w/ sync dispatcher?
+		this.environment = new Environment(new SynchronousDispatcherConfigReader());
+
+		this.tcpClient = new TcpClientSpec<Message<P>, Message<P>>(REACTOR_TCP_CLIENT_TYPE)
+				.env(this.environment)
+				.codec(codec)
+				.connect(host, port)
+				.get();
+	}
+
+	/**
+	 * A constructor with a pre-configured {@link reactor.net.tcp.TcpClient}.
+	 *
+	 * <p><strong>NOTE:</strong> if the client is configured with a thread-creating
+	 * dispatcher, you are responsible for shutting down the {@link reactor.core.Environment}
+	 * instance with which the client is configured.
+	 *
+	 * @param tcpClient the TcpClient to use
+	 */
+	public Reactor11TcpClient(TcpClient<Message<P>, Message<P>> tcpClient) {
+		Assert.notNull(tcpClient, "'tcpClient' must not be null");
+		this.tcpClient = tcpClient;
+		this.environment = null;
+	}
+
+
+	@Override
+	public ListenableFuture<Void> connect(TcpConnectionHandler<P> connectionHandler) {
+
+		Promise<NetChannel<Message<P>, Message<P>>> promise = this.tcpClient.open();
+		composeConnectionHandling(promise, connectionHandler);
+
+		return new AbstractPromiseToListenableFutureAdapter<NetChannel<Message<P>, Message<P>>, Void>(promise) {
+			@Override
+			protected Void adapt(NetChannel<Message<P>, Message<P>> result) {
+				return null;
+			}
+		};
+	}
+
+	@Override
+	public ListenableFuture<Void> connect(final TcpConnectionHandler<P> connectionHandler,
+			final ReconnectStrategy reconnectStrategy) {
+
+		Assert.notNull(reconnectStrategy, "ReconnectStrategy must not be null");
+
+		Reconnect reconnect = new Reconnect() {
+			@Override
+			public Tuple2<InetSocketAddress, Long> reconnect(InetSocketAddress address, int attempt) {
+				return Tuple.of(address, reconnectStrategy.getTimeToNextAttempt(attempt));
+			}
+		};
+
+		Stream<NetChannel<Message<P>, Message<P>>> stream = this.tcpClient.open(reconnect);
+		composeConnectionHandling(stream, connectionHandler);
+
+		Promise<Void> promise = Promises.next(stream).map(
+				new Function<NetChannel<Message<P>, Message<P>>, Void>() {
+					@Override
+					public Void apply(NetChannel<Message<P>, Message<P>> ch) {
+						return null;
+					}
+				});
+		return new PassThroughPromiseToListenableFutureAdapter<Void>(promise);
+	}
+
+	private void composeConnectionHandling(Composable<NetChannel<Message<P>, Message<P>>> composable,
+			final TcpConnectionHandler<P> connectionHandler) {
+
+		composable
+				.when(Throwable.class, new Consumer<Throwable>() {
+					@Override
+					public void accept(Throwable ex) {
+						connectionHandler.afterConnectFailure(ex);
+					}
+				})
+				.consume(new Consumer<NetChannel<Message<P>, Message<P>>>() {
+					@Override
+					public void accept(NetChannel<Message<P>, Message<P>> connection) {
+						connection
+								.when(Throwable.class, new Consumer<Throwable>() {
+									@Override
+									public void accept(Throwable t) {
+										connectionHandler.handleFailure(t);
+									}
+								})
+								.consume(new Consumer<Message<P>>() {
+									@Override
+									public void accept(Message<P> message) {
+										connectionHandler.handleMessage(message);
+									}
+								})
+								.on()
+								.close(new Runnable() {
+									@Override
+									public void run() {
+										connectionHandler.afterConnectionClosed();
+									}
+								});
+						connectionHandler.afterConnected(new Reactor11TcpConnection<P>(connection));
+					}
+				});
+	}
+
+	@Override
+	public ListenableFuture<Boolean> shutdown() {
+		try {
+			Promise<Boolean> promise = this.tcpClient.close();
+			return new AbstractPromiseToListenableFutureAdapter<Boolean, Boolean>(promise) {
+				@Override
+				protected Boolean adapt(Boolean result) {
+					return result;
+				}
+			};
+		}
+		finally {
+			this.environment.shutdown();
+
+		}
+	}
+
+	/**
+	 * A ConfigurationReader that enforces the use of a SynchronousDispatcher.
+	 *
+	 * <p>The {@link reactor.core.configuration.PropertiesConfigurationReader} used by
+	 * default automatically creates other dispatchers with thread pools that are
+	 * not needed.
+	 */
+	private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
+
+		@Override
+		public ReactorConfiguration read() {
+			return new ReactorConfiguration(Arrays.<DispatcherConfiguration>asList(), "sync", new Properties());
+		}
+	}
+
+}

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

@@ -0,0 +1,67 @@
+/*
+ * 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.tcp.reactor;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.tcp.TcpConnection;
+import org.springframework.messaging.tcp.reactor.PassThroughPromiseToListenableFutureAdapter;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import reactor.core.composable.Promise;
+import reactor.net.NetChannel;
+
+
+/**
+ * An implementation of {@link org.springframework.messaging.tcp.TcpConnection}
+ * based on the TCP client support of the Reactor project.
+ *
+ * @param <P> the payload type of Spring Message's read from
+ * and written to the TCP stream
+ *
+ * @author Rossen Stoyanchev
+ */
+public class Reactor11TcpConnection<P> implements TcpConnection<P> {
+
+	private final NetChannel<Message<P>, Message<P>> channel;
+
+
+	public Reactor11TcpConnection(NetChannel<Message<P>, Message<P>> connection) {
+		this.channel = connection;
+	}
+
+	@Override
+	public ListenableFuture<Void> send(Message<P> message) {
+		Promise<Void> promise = this.channel.send(message);
+		return new PassThroughPromiseToListenableFutureAdapter<Void>(promise);
+	}
+
+	@Override
+	public void onReadInactivity(Runnable runnable, long inactivityDuration) {
+		this.channel.on().readIdle(inactivityDuration, runnable);
+	}
+
+	@Override
+	public void onWriteInactivity(Runnable runnable, long inactivityDuration) {
+		this.channel.on().writeIdle(inactivityDuration, runnable);
+	}
+
+	@Override
+	public void close() {
+		this.channel.close();
+	}
+
+}

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

@@ -19,7 +19,6 @@ package org.springframework.messaging.tcp.reactor;
 import java.lang.reflect.Modifier;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Properties;
 
 import org.apache.commons.logging.Log;
@@ -217,13 +216,13 @@ public class ReactorTcpClient<P> implements TcpOperations<P> {
 	}
 
 	@Override
-	public ListenableFuture<Void> shutdown() {
+	public ListenableFuture<Boolean> shutdown() {
 		try {
 			Promise<Void> promise = this.tcpClient.close();
-			return new AbstractPromiseToListenableFutureAdapter<Void, Void>(promise) {
+			return new AbstractPromiseToListenableFutureAdapter<Void, Boolean>(promise) {
 				@Override
-				protected Void adapt(Void result) {
-					return result;
+				protected Boolean adapt(Void result) {
+					return true;
 				}
 			};
 		}

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java

@@ -19,18 +19,13 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
 import org.junit.Test;
-import org.springframework.context.ApplicationEvent;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.StubMessageChannel;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageType;
-import org.springframework.messaging.simp.broker.BrokerAvailabilityEvent;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.messaging.tcp.ReconnectStrategy;
 import org.springframework.messaging.tcp.TcpConnection;
@@ -150,6 +145,17 @@ public class StompBrokerRelayMessageHandlerTests {
 		return futureTask;
 	}
 
+	private static ListenableFutureTask<Boolean> getBooleanFuture() {
+		ListenableFutureTask<Boolean> futureTask = new ListenableFutureTask<>(new Callable<Boolean>() {
+			@Override
+			public Boolean call() throws Exception {
+				return true;
+			}
+		});
+		futureTask.run();
+		return futureTask;
+	}
+
 
 	private static class StubTcpOperations implements TcpOperations<byte[]> {
 
@@ -169,8 +175,8 @@ public class StompBrokerRelayMessageHandlerTests {
 		}
 
 		@Override
-		public ListenableFuture<Void> shutdown() {
-			return getFuture();
+		public ListenableFuture<Boolean> shutdown() {
+			return getBooleanFuture();
 		}
 	}