Merge pull request #1250 from origin/update-stomp-reactor-netty

Author: rstoyanchev

build.gradle

@@ -78,7 +78,6 @@ configure(allprojects) { project ->
 	ext.protobufVersion        = "3.1.0"
 	ext.quartzVersion          = "2.2.3"
 	ext.reactivestreamsVersion = "1.0.0"
-	ext.reactorVersion         = "2.0.8.RELEASE"
 	ext.reactorCoreVersion     = '3.0.3.RELEASE'
 	ext.reactorNettyVersion    = '0.6.0.BUILD-SNAPSHOT'
 	ext.romeVersion            = "1.7.0"
@@ -578,12 +577,8 @@ project("spring-messaging") {
 		compile(project(":spring-core"))
 		compile(project(":spring-context"))
 		optional(project(":spring-oxm"))
-		optional("io.projectreactor:reactor-core:${reactorVersion}") {
-			force = true  // enforce 2.0.x
-		}
-		optional("io.projectreactor:reactor-net:${reactorVersion}") {
-			exclude group: "io.netty", module: "netty-all"
-		}
+		optional("io.projectreactor:reactor-core:${reactorCoreVersion}")
+		optional("io.projectreactor.ipc:reactor-netty:${reactorNettyVersion}")
 		optional("io.netty:netty-all:${nettyVersion}")
 		optional("org.eclipse.jetty.websocket:websocket-server:${jettyVersion}") {
 			exclude group: "javax.servlet", module: "javax.servlet-api"
@@ -1003,10 +998,8 @@ project("spring-websocket") {
 		optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}")
 		testCompile("org.apache.tomcat.embed:tomcat-embed-core:${tomcatVersion}")
 		testCompile("org.apache.tomcat.embed:tomcat-embed-websocket:${tomcatVersion}")
-		testCompile("io.projectreactor:reactor-core:${reactorVersion}") {
-			force = true  // enforce 2.0.x
-		}
-		testCompile("io.projectreactor:reactor-net:${reactorVersion}")
+		testCompile("io.projectreactor:reactor-core:${reactorCoreVersion}")
+		testCompile("io.projectreactor.ipc:reactor-netty:${reactorNettyVersion}")
 		testCompile("io.netty:netty-all:${nettyVersion}")
 		testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}")
 		testRuntime("org.jboss.xnio:xnio-nio:${xnioVersion}")

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

@@ -0,0 +1,91 @@
+/*
+ * 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.
+ * 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.tcp.TcpOperations;
+import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient;
+import org.springframework.util.Assert;
+import org.springframework.util.concurrent.ListenableFuture;
+
+/**
+ * A STOMP over TCP client that uses {@link ReactorNettyTcpClient}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+public class ReactorNettyTcpStompClient extends StompClientSupport {
+
+	private final TcpOperations<byte[]> tcpClient;
+
+
+	/**
+	 * Create an instance with host "127.0.0.1" and port 61613.
+	 */
+	public ReactorNettyTcpStompClient() {
+		this("127.0.0.1", 61613);
+	}
+
+	/**
+	 * Create an instance with the given host and port.
+	 * @param host the host
+	 * @param port the port
+	 */
+	public ReactorNettyTcpStompClient(String host, int port) {
+		this.tcpClient = new ReactorNettyTcpClient<byte[]>(host, port, new StompReactorNettyCodec());
+	}
+
+	/**
+	 * Create an instance with a pre-configured TCP client.
+	 * @param tcpClient the client to use
+	 */
+	public ReactorNettyTcpStompClient(TcpOperations<byte[]> tcpClient) {
+		Assert.notNull(tcpClient, "'tcpClient' is required");
+		this.tcpClient = tcpClient;
+	}
+
+	/**
+	 * Connect and notify the given {@link StompSessionHandler} when connected
+	 * on the STOMP level.
+	 * @param handler the handler for the STOMP session
+	 * @return ListenableFuture for access to the session when ready for use
+	 */
+	public ListenableFuture<StompSession> connect(StompSessionHandler handler) {
+		return connect(null, handler);
+	}
+
+
+	/**
+	 * An overloaded version of {@link #connect(StompSessionHandler)} that
+	 * accepts headers to use for the STOMP CONNECT frame.
+	 * @param connectHeaders headers to add to the CONNECT frame
+	 * @param handler the handler for the STOMP session
+	 * @return ListenableFuture for access to the session when ready for use
+	 */
+	public ListenableFuture<StompSession> connect(StompHeaders connectHeaders, StompSessionHandler handler) {
+		ConnectionHandlingStompSession session = createSession(connectHeaders, handler);
+		this.tcpClient.connect(session);
+		return session.getSessionFuture();
+	}
+
+	/**
+	 * Shut down the client and release resources.
+	 */
+	public void shutdown() {
+		this.tcpClient.shutdown();
+	}
+
+}