Polishing

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

@@ -20,14 +20,15 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import reactor.fn.Consumer;
+import reactor.rx.Promise;
+
 import org.springframework.util.Assert;
 import org.springframework.util.concurrent.FailureCallback;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
 import org.springframework.util.concurrent.ListenableFutureCallbackRegistry;
 import org.springframework.util.concurrent.SuccessCallback;
-import reactor.fn.Consumer;
-import reactor.rx.Promise;
 
 /**
  * Adapts a reactor {@link Promise} to {@link ListenableFuture} optionally converting
@@ -55,21 +56,20 @@ abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements Listena
 				try {
 					registry.success(adapt(result));
 				}
-				catch (Throwable t) {
-					registry.failure(t);
+				catch (Throwable ex) {
+					registry.failure(ex);
 				}
 			}
 		});
 
 		this.promise.onError(new Consumer<Throwable>() {
 			@Override
-			public void accept(Throwable t) {
-				registry.failure(t);
+			public void accept(Throwable ex) {
+				registry.failure(ex);
 			}
 		});
 	}
 
-	protected abstract T adapt(S result);
 
 	@Override
 	public T get() throws InterruptedException {
@@ -112,4 +112,7 @@ abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements Listena
 		this.registry.addFailureCallback(failureCallback);
 	}
 
+
+	protected abstract T adapt(S result);
+
 }

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

@@ -60,7 +60,6 @@ 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.ClassUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.concurrent.ListenableFuture;
 
@@ -83,7 +82,6 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	private static final Method eventLoopGroupMethod = initEventLoopGroupMethod();
 
 
-
 	private final EventLoopGroup eventLoopGroup;
 
 	private final TcpClientFactory<Message<P>, Message<P>> tcpClientSpecFactory;
@@ -107,7 +105,6 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	 * @param codec the codec to use for encoding and decoding the TCP stream
 	 */
 	public Reactor2TcpClient(final String host, final int port, final Codec<Buffer, Message<P>, Message<P>> codec) {
-
 		// Reactor 2.0.5 requires NioEventLoopGroup vs 2.0.6+ requires EventLoopGroup
 		final NioEventLoopGroup nioEventLoopGroup = initEventLoopGroup();
 		this.eventLoopGroup = nioEventLoopGroup;
@@ -182,8 +179,8 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 		return new PassThroughPromiseToListenableFutureAdapter<Void>(
 				promise.onError(new Consumer<Throwable>() {
 					@Override
-					public void accept(Throwable throwable) {
-						connectionHandler.afterConnectFailure(throwable);
+					public void accept(Throwable ex) {
+						connectionHandler.afterConnectFailure(ex);
 					}
 				})
 		);
@@ -262,7 +259,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 				return method;
 			}
 		}
-		throw new IllegalStateException("No compatible Reactor version found.");
+		throw new IllegalStateException("No compatible Reactor version found");
 	}
 
 
@@ -270,8 +267,8 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 
 		@Override
 		public ReactorConfiguration read() {
-			return new ReactorConfiguration(Collections.<DispatcherConfiguration>emptyList(),
-					"sync", new Properties());
+			return new ReactorConfiguration(
+					Collections.<DispatcherConfiguration>emptyList(), "sync", new Properties());
 		}
 	}