Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4125: Fix STOMP module to the latest SF

Browse Source 
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
This commit is contained in:
Artem Bilan
2016-12-02 12:21:43 -05:00
parent 7ddfe49b8b
commit b25c95508e
3 changed files with 27 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
build.gradle
View File

@@ -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"
}
}

18
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java → spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java
View File

@@ -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<StompSession> doConnect(StompSessionHandler handler) {
return ((Reactor2TcpStompClient) this.stompClient).connect(getConnectHeaders(), handler);
return ((ReactorNettyTcpStompClient) this.stompClient).connect(getConnectHeaders(), handler);
}
}

15
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
View File

@@ -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;