Polish Reactor Netty TCP client support

Author: rstoyanchev

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

@@ -29,11 +29,10 @@
 import org.springframework.util.concurrent.ListenableFuture;
 
 /**
- * A STOMP over TCP client that uses
- * {@link ReactorNettyTcpClient}.
+ * A STOMP over TCP client that uses {@link ReactorNettyTcpClient}.
  *
  * @author Rossen Stoyanchev
- * @since 4.2
+ * @since 5.0
  */
 public class ReactorNettyTcpStompClient extends StompClientSupport {
 
@@ -99,25 +98,21 @@ public void shutdown() {
 	 * Create a new {@link ReactorNettyTcpClient} with Stomp specific configuration for
 	 * encoding, decoding and hand-off.
 	 *
-	 * @param relayHost target host
-	 * @param relayPort target port
+	 * @param host target host
+	 * @param port target port
 	 * @param decoder {@link StompDecoder} to use
 	 * @return a new {@link TcpOperations}
 	 */
-	protected static TcpOperations<byte[]> create(String relayHost,
-			int relayPort,
-			StompDecoder decoder) {
-		return new ReactorNettyTcpClient<>(relayHost,
-				relayPort,
-				new ReactorNettyTcpClient.MessageHandlerConfiguration<>(new DecodingFunction(
-						decoder),
+	protected static TcpOperations<byte[]> create(String host, int port, StompDecoder decoder) {
+		return new ReactorNettyTcpClient<>(host, port,
+				new ReactorNettyTcpClient.MessageHandlerConfiguration<>(
+						new DecodingFunction(decoder),
 						new EncodingConsumer(new StompEncoder()),
 						128,
 						Schedulers.newParallel("StompClient")));
 	}
 
-	private static final class EncodingConsumer
-			implements BiConsumer<ByteBuf, Message<byte[]>> {
+	private static final class EncodingConsumer implements BiConsumer<ByteBuf, Message<byte[]>> {
 
 		private final StompEncoder encoder;
 
@@ -127,12 +122,11 @@ public EncodingConsumer(StompEncoder encoder) {
 
 		@Override
 		public void accept(ByteBuf byteBuf, Message<byte[]> message) {
-			byteBuf.writeBytes(encoder.encode(message));
+			byteBuf.writeBytes(this.encoder.encode(message));
 		}
 	}
 
-	private static final class DecodingFunction
-			implements Function<ByteBuf, List<Message<byte[]>>> {
+	private static final class DecodingFunction implements Function<ByteBuf, List<Message<byte[]>>> {
 
 		private final StompDecoder decoder;
 

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

@@ -17,7 +17,6 @@
 package org.springframework.messaging.tcp.reactor;
 
 import java.time.Duration;
-import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -33,51 +32,53 @@
 import org.springframework.util.concurrent.SuccessCallback;
 
 /**
- * Adapts a reactor {@link Mono} to {@link ListenableFuture} optionally converting
- * the result Object type {@code <S>} to the expected target type {@code <T>}.
+ * Adapts {@link Mono} to {@link ListenableFuture} optionally converting the
+ * result Object type {@code <S>} to the expected target type {@code <T>}.
  *
  * @author Rossen Stoyanchev
- * @since 4.0
+ * @since 5.0
  * @param <S> the type of object expected from the {@link Mono}
  * @param <T> the type of object expected from the {@link ListenableFuture}
  */
-abstract class AbstractMonoToListenableFutureAdapter<S, T>
-		implements ListenableFuture<T> {
+abstract class AbstractMonoToListenableFutureAdapter<S, T> implements ListenableFuture<T> {
 
-	private final MonoProcessor<S> promise;
+	private final MonoProcessor<S> monoProcessor;
 
 	private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<>();
 
-	protected AbstractMonoToListenableFutureAdapter(Mono<S> promise) {
-		Assert.notNull(promise, "Mono must not be null");
-		this.promise = promise.doOnSuccess(result -> {
-				T adapted;
-				try {
-					adapted = adapt(result);
-				}
-				catch (Throwable ex) {
-					registry.failure(ex);
-					return;
-				}
-				registry.success(adapted);
-		})
-		                      .doOnError(registry::failure)
-		                      .subscribe();
+
+	protected AbstractMonoToListenableFutureAdapter(Mono<S> mono) {
+		Assert.notNull(mono, "'mono' must not be null");
+		this.monoProcessor = mono
+				.doOnSuccess(result -> {
+					T adapted;
+					try {
+						adapted = adapt(result);
+					}
+					catch (Throwable ex) {
+						registry.failure(ex);
+						return;
+					}
+					registry.success(adapted);
+				})
+				.doOnError(this.registry::failure)
+				.subscribe();
 	}
 
 
 	@Override
 	public T get() throws InterruptedException {
-		S result = this.promise.block();
+		S result = this.monoProcessor.block();
 		return adapt(result);
 	}
 
 	@Override
-	public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-		Objects.requireNonNull(unit, "unit");
-		S result = this.promise.block(Duration.ofMillis(TimeUnit.MILLISECONDS.convert(
-				timeout,
-				unit)));
+	public T get(long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+
+		Assert.notNull(unit);
+		Duration duration = Duration.ofMillis(TimeUnit.MILLISECONDS.convert(timeout, unit));
+		S result = this.monoProcessor.block(duration);
 		return adapt(result);
 	}
 
@@ -86,18 +87,18 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 		if (isCancelled()) {
 			return false;
 		}
-		this.promise.cancel();
+		this.monoProcessor.cancel();
 		return true;
 	}
 
 	@Override
 	public boolean isCancelled() {
-		return this.promise.isCancelled();
+		return this.monoProcessor.isCancelled();
 	}
 
 	@Override
 	public boolean isDone() {
-		return this.promise.isTerminated();
+		return this.monoProcessor.isTerminated();
 	}
 
 	@Override
@@ -111,7 +112,6 @@ public void addCallback(SuccessCallback<? super T> successCallback, FailureCallb
 		this.registry.addFailureCallback(failureCallback);
 	}
 
-
 	protected abstract T adapt(S result);
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -19,15 +19,14 @@
 import reactor.core.publisher.Mono;
 
 /**
- * A Mono-to-ListenableFutureAdapter where the source and the target from
+ * A Mono-to-ListenableFuture adapter where the source and the target from
  * the Promise and the ListenableFuture respectively are of the same type.
  *
  * @author Rossen Stoyanchev
  * @author Stephane Maldini
- * @since 4.0
+ * @since 5.0
  */
-class MonoToListenableFutureAdapter<T> extends
-                                       AbstractMonoToListenableFutureAdapter<T, T> {
+class MonoToListenableFutureAdapter<T> extends AbstractMonoToListenableFutureAdapter<T, T> {
 
 
 	public MonoToListenableFutureAdapter(Mono<T> mono) {