From 8c316e1863212c97c8d58e7de78fbfd85401660a Mon Sep 17 00:00:00 2001
From: Rossen Stoyanchev <rstoyanchev@pivotal.io>
Date: Wed, 30 Sep 2015 18:41:29 -0400
Subject: [PATCH] Avoid issue with switching from Reactor 2.0.5 to 2.0.6

We can't compile directly against NettyClientSocketOptions method which
changed signatures in 2.0.6. This change ensures the method is invoked
reflectively instead.
---
 .../tcp/reactor/Reactor2TcpClient.java        | 27 ++++++++++++++++---
 1 file changed, 24 insertions(+), 3 deletions(-)

diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java
index ceb563b8c2..1e8976ba78 100644
--- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java
+++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java
@@ -16,9 +16,9 @@
 
 package org.springframework.messaging.tcp.reactor;
 
+import java.lang.reflect.Method;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
@@ -45,6 +45,7 @@ import reactor.io.net.NetStreams.TcpClientFactory;
 import reactor.io.net.ReactorChannelHandler;
 import reactor.io.net.Reconnect;
 import reactor.io.net.Spec.TcpClientSpec;
+import reactor.io.net.config.ClientSocketOptions;
 import reactor.io.net.impl.netty.NettyClientSocketOptions;
 import reactor.io.net.impl.netty.tcp.NettyTcpClient;
 import reactor.io.net.tcp.TcpClient;
@@ -59,6 +60,8 @@ 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;
 
 /**
@@ -77,6 +80,9 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	@SuppressWarnings("rawtypes")
 	public static final Class<NettyTcpClient> REACTOR_TCP_CLIENT_TYPE = NettyTcpClient.class;
 
+	private static final Method eventLoopGroupMethod = initEventLoopGroupMethod();
+
+
 
 	private final EventLoopGroup eventLoopGroup;
 
@@ -102,7 +108,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	 */
 	public Reactor2TcpClient(final String host, final int port, final Codec<Buffer, Message<P>, Message<P>> codec) {
 
-		// Reactor 2.0.5 required NioEventLoopGroup (2.0.6 changed to EventLoopGroup)
+		// Reactor 2.0.5 requires NioEventLoopGroup vs 2.0.6+ requires EventLoopGroup
 		final NioEventLoopGroup nioEventLoopGroup = initEventLoopGroup();
 		this.eventLoopGroup = nioEventLoopGroup;
 
@@ -113,7 +119,12 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 						.env(new Environment(new SynchronousDispatcherConfigReader()))
 						.codec(codec)
 						.connect(host, port)
-						.options(new NettyClientSocketOptions().eventLoopGroup(nioEventLoopGroup));
+						.options(createClientSocketOptions());
+			}
+
+			private ClientSocketOptions createClientSocketOptions() {
+				return (ClientSocketOptions) ReflectionUtils.invokeMethod(eventLoopGroupMethod,
+						new NettyClientSocketOptions(), nioEventLoopGroup);
 			}
 		};
 	}
@@ -245,6 +256,16 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	}
 
 
+	private static Method initEventLoopGroupMethod() {
+		for (Method method : NettyClientSocketOptions.class.getMethods()) {
+			if (method.getName().equals("eventLoopGroup") && method.getParameterTypes().length == 1) {
+				return method;
+			}
+		}
+		throw new IllegalStateException("No compatible Reactor version found.");
+	}
+
+
 	private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
 
 		@Override