From b25c95508e946a6f5061b897167b0fb0e44fdb88 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 2 Dec 2016 12:21:43 -0500 Subject: [PATCH] INT-4125: Fix STOMP module to the latest SF JIRA: https://jira.spring.io/browse/INT-4125 * Rename `Reactor2TcpStompSessionManager` to `ReactorNettyTcpStompSessionManager` and make it based on the `ReactorNettyTcpStompClient` from the latest SF * Revert dependencies to their latest B-S --- build.gradle | 21 +++++++++---------- ...> ReactorNettyTcpStompSessionManager.java} | 18 +++++++++------- .../client/StompServerIntegrationTests.java | 15 +++++++------ 3 files changed, 27 insertions(+), 27 deletions(-) rename spring-integration-stomp/src/main/java/org/springframework/integration/stomp/{Reactor2TcpStompSessionManager.java => ReactorNettyTcpStompSessionManager.java} (64%) diff --git a/build.gradle b/build.gradle index a33e1a2b00..ee36c8ff34 100644 --- a/build.gradle +++ b/build.gradle @@ -113,9 +113,9 @@ subprojects { subproject -> log4jVersion = '1.2.17' mockitoVersion = '1.10.19' mysqlVersion = '5.1.34' - nettyVersion = '4.1.4.Final' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' + reactorNettyVersion = '0.6.0.BUILD-SNAPSHOT' reactorVersion = '3.0.3.RELEASE' romeToolsVersion = '1.7.0' servletApiVersion = '3.1.0' @@ -123,14 +123,14 @@ subprojects { subproject -> tomcatVersion = "8.0.33" smackVersion = '4.1.7' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.0.0.BUILD-SNAPSHOT' - springDataJpaVersion = '2.0.0.M1' - springDataMongoVersion = '2.0.0.M1' - springDataRedisVersion = '2.0.0.M1' - springGemfireVersion = '2.0.0.M1' + springDataJpaVersion = '2.0.0.BUILD-SNAPSHOT' + springDataMongoVersion = '2.0.0.BUILD-SNAPSHOT' + springDataRedisVersion = '2.0.0.BUILD-SNAPSHOT' + springGemfireVersion = '2.0.0.BUILD-SNAPSHOT' springSecurityVersion = '4.2.0.RELEASE' springSocialTwitterVersion = '1.1.2.RELEASE' springRetryVersion = '1.2.0.RC1' - springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.0.M3' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.0.BUILD-SNAPSHOT' springWsVersion = '2.4.0.RELEASE' xmlUnitVersion = '1.6' xstreamVersion = '1.4.7' @@ -515,6 +515,8 @@ project('spring-integration-redis') { exclude group: 'org.springframework', module: 'spring-context-support' exclude group: 'org.springframework', module: 'spring-core' exclude group: 'org.springframework', module: 'spring-tx' + exclude group: 'org.springframework', module: 'spring-aop' + exclude group: 'org.springframework', module: 'spring-oxm' } testCompile "redis.clients:jedis:$jedisVersion" } @@ -569,6 +571,8 @@ project('spring-integration-stomp') { compile ("org.springframework:spring-websocket:$springVersion", optional) + compile ("io.projectreactor.ipc:reactor-netty:$reactorNettyVersion" , optional) + testCompile project(":spring-integration-websocket") testCompile project(":spring-integration-websocket").sourceSets.test.output testCompile project(":spring-integration-event") @@ -576,11 +580,6 @@ project('spring-integration-stomp') { testCompile "org.apache.tomcat.embed:tomcat-embed-websocket:$tomcatVersion" testRuntime "org.apache.tomcat.embed:tomcat-embed-logging-log4j:$tomcatVersion" - testRuntime("io.projectreactor:reactor-core:2.0.8.RELEASE") { - force = true // enforce 2.0.x - } - testRuntime("io.projectreactor:reactor-net:2.0.8.RELEASE") - testRuntime "io.netty:netty-all:$nettyVersion" } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java similarity index 64% rename from spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java rename to spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java index 2da1386691..e19baefedb 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java @@ -16,27 +16,29 @@ package org.springframework.integration.stomp; -import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient; +import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient; import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; import org.springframework.util.concurrent.ListenableFuture; /** - * The {@link Reactor2TcpStompClient} based {@link AbstractStompSessionManager} implementation. + * The {@link ReactorNettyTcpStompClient} based {@link AbstractStompSessionManager} implementation. * * @author Artem Bilan - * @see Reactor2TcpStompClient - * @since 4.2 + * + * @since 5.0 + * + * @see ReactorNettyTcpStompClient */ -public class Reactor2TcpStompSessionManager extends AbstractStompSessionManager { +public class ReactorNettyTcpStompSessionManager extends AbstractStompSessionManager { - public Reactor2TcpStompSessionManager(Reactor2TcpStompClient reactor2TcpStompClient) { - super(reactor2TcpStompClient); + public ReactorNettyTcpStompSessionManager(ReactorNettyTcpStompClient reactorNettyTcpStompClient) { + super(reactorNettyTcpStompClient); } @Override protected ListenableFuture doConnect(StompSessionHandler handler) { - return ((Reactor2TcpStompClient) this.stompClient).connect(getConnectHeaders(), handler); + return ((ReactorNettyTcpStompClient) this.stompClient).connect(getConnectHeaders(), handler); } } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java index b450596d90..b994f12eff 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java @@ -28,7 +28,6 @@ import static org.junit.Assert.fail; import org.apache.activemq.broker.BrokerService; import org.junit.AfterClass; import org.junit.BeforeClass; -import org.junit.Ignore; import org.junit.Test; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; @@ -41,7 +40,8 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer; -import org.springframework.integration.stomp.Reactor2TcpStompSessionManager; +import org.springframework.integration.stomp.AbstractStompSessionManager; +import org.springframework.integration.stomp.ReactorNettyTcpStompSessionManager; import org.springframework.integration.stomp.StompSessionManager; import org.springframework.integration.stomp.event.StompConnectionFailedEvent; import org.springframework.integration.stomp.event.StompIntegrationEvent; @@ -56,7 +56,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; -import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient; +import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -67,17 +67,16 @@ import org.springframework.util.SocketUtils; * @author Gary Russell * @since 4.2 */ -@Ignore("Until Reactor 3.0 support: https://jira.spring.io/browse/INT-4125. IO Cairo doesn't support Reactor 2.0") public class StompServerIntegrationTests extends LogAdjustingTestSupport { private static BrokerService activeMQBroker; - private static Reactor2TcpStompClient stompClient; + private static ReactorNettyTcpStompClient stompClient; public StompServerIntegrationTests() { super("org.springframework", "org.springframework.integration.stomp", - "org.apache.activemq.broker", "reactor.io", "io.netty"); + "org.apache.activemq.broker", "reactor.ipc", "io.netty"); } @BeforeClass @@ -91,7 +90,7 @@ public class StompServerIntegrationTests extends LogAdjustingTestSupport { activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5); activeMQBroker.start(); - stompClient = new Reactor2TcpStompClient("127.0.0.1", port); + stompClient = new ReactorNettyTcpStompClient("127.0.0.1", port); stompClient.setMessageConverter(new PassThruMessageConverter()); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); @@ -242,7 +241,7 @@ public class StompServerIntegrationTests extends LogAdjustingTestSupport { @Bean public StompSessionManager stompSessionManager() { - Reactor2TcpStompSessionManager stompSessionManager = new Reactor2TcpStompSessionManager(stompClient); + AbstractStompSessionManager stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient); stompSessionManager.setAutoReceipt(true); stompSessionManager.setRecoveryInterval(500); return stompSessionManager;