diff --git a/build.gradle b/build.gradle index 31872face7..0fa702ed9b 100644 --- a/build.gradle +++ b/build.gradle @@ -75,6 +75,7 @@ configure(allprojects) { project -> } repositories { + maven { url "http://repo.spring.io/snapshot" } // temporarily until Reactor 1.1.0.M4 maven { url "http://repo.spring.io/libs-release" } maven { url "http://repo.spring.io/milestone" } // for AspectJ 1.8.0.M1 maven { url "https://oss.sonatype.org/content/repositories/releases" } // javax.cache @@ -398,8 +399,8 @@ project("spring-messaging") { compile(project(":spring-beans")) compile(project(":spring-core")) compile(project(":spring-context")) - optional("org.projectreactor:reactor-core:1.0.1.RELEASE") - optional("org.projectreactor:reactor-tcp:1.0.1.RELEASE") + optional("org.projectreactor:reactor-core:1.1.0.BUILD-SNAPSHOT") + optional("org.projectreactor:reactor-net:1.1.0.BUILD-SNAPSHOT") optional("org.eclipse.jetty.websocket:websocket-server:${jettyVersion}") { exclude group: "javax.servlet", module: "javax.servlet-api" } @@ -621,8 +622,8 @@ project("spring-websocket") { testCompile("org.apache.tomcat.embed:tomcat-embed-core:8.0.3") testCompile("org.apache.tomcat.embed:tomcat-embed-websocket:8.0.3") testCompile("org.apache.tomcat.embed:tomcat-embed-logging-juli:8.0.3") - testCompile("org.projectreactor:reactor-core:1.0.1.RELEASE") - testCompile("org.projectreactor:reactor-tcp:1.0.1.RELEASE") + testCompile("org.projectreactor:reactor-core:1.1.0.BUILD-SNAPSHOT") + testCompile("org.projectreactor:reactor-net:1.1.0.BUILD-SNAPSHOT") testCompile("log4j:log4j:1.2.17") testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}") } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCodec.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCodec.java index 1425860bdc..ec5569508f 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCodec.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCodec.java @@ -21,7 +21,7 @@ import org.springframework.messaging.Message; import reactor.function.Consumer; import reactor.function.Function; import reactor.io.Buffer; -import reactor.tcp.encoding.Codec; +import reactor.io.encoding.Codec; import java.util.List; diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpOperations.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpOperations.java index 6070ad13d6..8db2e10140 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpOperations.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/TcpOperations.java @@ -50,6 +50,6 @@ public interface TcpOperations

{ * @return a ListenableFuture that can be used to determine when and if the * connection is successfully closed */ - ListenableFuture shutdown(); + ListenableFuture shutdown(); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpClient.java index c58bbffe3d..8722449d54 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpClient.java @@ -16,10 +16,8 @@ package org.springframework.messaging.tcp.reactor; -import java.lang.reflect.Modifier; import java.net.InetSocketAddress; import java.util.Arrays; -import java.util.Collections; import java.util.Properties; import org.apache.commons.logging.Log; @@ -33,7 +31,6 @@ import org.springframework.util.concurrent.ListenableFuture; import reactor.core.Environment; import reactor.core.composable.Composable; -import reactor.core.composable.Deferred; import reactor.core.composable.Promise; import reactor.core.composable.Stream; import reactor.core.composable.spec.Promises; @@ -41,14 +38,14 @@ import reactor.core.configuration.ConfigurationReader; import reactor.core.configuration.DispatcherConfiguration; import reactor.core.configuration.ReactorConfiguration; import reactor.function.Consumer; -import reactor.function.support.SingleUseConsumer; +import reactor.function.Function; import reactor.io.Buffer; -import reactor.tcp.Reconnect; -import reactor.tcp.TcpClient; -import reactor.tcp.TcpConnection; -import reactor.tcp.encoding.Codec; -import reactor.tcp.netty.NettyTcpClient; -import reactor.tcp.spec.TcpClientSpec; +import reactor.io.encoding.Codec; +import reactor.net.NetChannel; +import reactor.net.Reconnect; +import reactor.net.netty.tcp.NettyTcpClient; +import reactor.net.tcp.TcpClient; +import reactor.net.tcp.spec.TcpClientSpec; import reactor.tuple.Tuple; import reactor.tuple.Tuple2; @@ -73,12 +70,12 @@ public class ReactorTcpClient

implements TcpOperations

{ /** - * A constructor that creates a {@link reactor.tcp.netty.NettyTcpClient} with + * A constructor that creates a {@link reactor.net.netty.tcp.NettyTcpClient} with * a {@link reactor.event.dispatch.SynchronousDispatcher} as a result of which * network I/O is handled in Netty threads. * *

Also see the constructor accepting a pre-configured Reactor - * {@link reactor.tcp.TcpClient}. + * {@link reactor.net.tcp.TcpClient}. * * @param host the host to connect to * @param port the port to connect to @@ -94,12 +91,10 @@ public class ReactorTcpClient

implements TcpOperations

{ .codec(codec) .connect(host, port) .get(); - - checkReactorVersion(); } /** - * A constructor with a pre-configured {@link reactor.tcp.TcpClient}. + * A constructor with a pre-configured {@link reactor.net.tcp.TcpClient}. * *

NOTE: if the client is configured with a thread-creating * dispatcher, you are responsible for shutting down the {@link reactor.core.Environment} @@ -111,31 +106,18 @@ public class ReactorTcpClient

implements TcpOperations

{ Assert.notNull(tcpClient, "'tcpClient' must not be null"); this.tcpClient = tcpClient; this.environment = null; - checkReactorVersion(); - } - - private static void checkReactorVersion() { - Class type = null; - try { - type = ReactorTcpClient.class.getClassLoader().loadClass("reactor.event.dispatch.BaseDispatcher"); - Assert.isTrue(Modifier.isPublic(type.getModifiers()), - "Detected older version of reactor-tcp. Switch to 1.0.1.RELEASE or higher."); - } - catch (ClassNotFoundException e) { - // Ignore, must be 1.1+ - } } @Override public ListenableFuture connect(TcpConnectionHandler

connectionHandler) { - Promise, Message

>> promise = this.tcpClient.open(); + Promise, Message

>> promise = this.tcpClient.open(); composeConnectionHandling(promise, connectionHandler); - return new AbstractPromiseToListenableFutureAdapter, Message

>, Void>(promise) { + return new AbstractPromiseToListenableFutureAdapter, Message

>, Void>(promise) { @Override - protected Void adapt(TcpConnection, Message

> result) { + protected Void adapt(NetChannel, Message

> result) { return null; } }; @@ -147,82 +129,71 @@ public class ReactorTcpClient

implements TcpOperations

{ Assert.notNull(reconnectStrategy, "ReconnectStrategy must not be null"); - Stream, Message

>> stream = - this.tcpClient.open(new Reconnect() { - @Override - public Tuple2 reconnect(InetSocketAddress address, int attempt) { - return Tuple.of(address, reconnectStrategy.getTimeToNextAttempt(attempt)); - } - }); + Reconnect reconnect = new Reconnect() { + @Override + public Tuple2 reconnect(InetSocketAddress address, int attempt) { + return Tuple.of(address, reconnectStrategy.getTimeToNextAttempt(attempt)); + } + }; + + Stream, Message

>> stream = this.tcpClient.open(reconnect); composeConnectionHandling(stream, connectionHandler); - return new PassThroughPromiseToListenableFutureAdapter(toPromise(stream)); + Promise promise = Promises.next(stream).map( + new Function, Message

>, Void>() { + @Override + public Void apply(NetChannel, Message

> ch) { + return null; + } + }); + return new PassThroughPromiseToListenableFutureAdapter(promise); } - private void composeConnectionHandling(Composable, Message

>> composable, + private void composeConnectionHandling(Composable, Message

>> composable, final TcpConnectionHandler

connectionHandler) { - composable.when(Throwable.class, new Consumer() { - @Override - public void accept(Throwable ex) { - connectionHandler.afterConnectFailure(ex); - } - }); - - composable.consume(new Consumer, Message

>>() { - @Override - public void accept(TcpConnection, Message

> connection) { - connection.on().close(new Runnable() { + composable + .when(Throwable.class, new Consumer() { @Override - public void run() { - connectionHandler.afterConnectionClosed(); + public void accept(Throwable ex) { + connectionHandler.afterConnectFailure(ex); + } + }) + .consume(new Consumer, Message

>>() { + @Override + public void accept(NetChannel, Message

> connection) { + connection + .when(Throwable.class, new Consumer() { + @Override + public void accept(Throwable t) { + logger.error("Exception on connection " + connectionHandler, t); + } + }) + .consume(new Consumer>() { + @Override + public void accept(Message

message) { + connectionHandler.handleMessage(message); + } + }) + .on() + .close(new Runnable() { + @Override + public void run() { + connectionHandler.afterConnectionClosed(); + } + }); + connectionHandler.afterConnected(new ReactorTcpConnection

(connection)); } }); - connection.consume(new Consumer>() { - @Override - public void accept(Message

message) { - connectionHandler.handleMessage(message); - } - }); - connection.when(Throwable.class, new Consumer() { - @Override - public void accept(Throwable t) { - logger.error("Exception on connection " + connectionHandler, t); - } - }); - connectionHandler.afterConnected(new ReactorTcpConnection

(connection)); - } - }); - } - - private Promise toPromise(Stream, Message

>> stream) { - - final Deferred> deferred = Promises.defer().get(); - - stream.consume(SingleUseConsumer.once(new Consumer, Message

>>() { - @Override - public void accept(TcpConnection, Message

> conn) { - deferred.accept((Void) null); - } - })); - - stream.when(Throwable.class, SingleUseConsumer.once(new Consumer() { - @Override - public void accept(Throwable throwable) { - deferred.accept(throwable); - } - })); - - return deferred.compose(); } @Override - public ListenableFuture shutdown() { + public ListenableFuture shutdown() { try { - Promise promise = this.tcpClient.close(); - return new AbstractPromiseToListenableFutureAdapter(promise) { + Promise promise = this.tcpClient.close(); + return new AbstractPromiseToListenableFutureAdapter(promise) { @Override - protected Void adapt(Void result) { + protected Boolean adapt(Boolean result) { return result; } }; @@ -233,7 +204,6 @@ public class ReactorTcpClient

implements TcpOperations

{ } } - /** * A ConfigurationReader that enforces the use of a SynchronousDispatcher. * diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpConnection.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpConnection.java index e4c47c9795..78033c865f 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpConnection.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorTcpConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -21,35 +21,46 @@ import org.springframework.messaging.tcp.TcpConnection; import org.springframework.util.concurrent.ListenableFuture; import reactor.core.composable.Promise; +import reactor.net.NetChannel; + +/** + * An implementation of {@link org.springframework.messaging.tcp.TcpConnection} + * based on the TCP client support of the Reactor project. + * + * @param

the payload type of Spring Message's read from + * and written to the TCP stream + * + * @author Rossen Stoyanchev + */ public class ReactorTcpConnection

implements TcpConnection

{ - private final reactor.tcp.TcpConnection, Message

> reactorTcpConnection; + private final NetChannel, Message

> channel; - public ReactorTcpConnection(reactor.tcp.TcpConnection, Message

> connection) { - this.reactorTcpConnection = connection; + public ReactorTcpConnection(NetChannel, Message

> connection) { + this.channel = connection; } @Override public ListenableFuture send(Message

message) { - Promise promise = this.reactorTcpConnection.send(message); + Promise promise = this.channel.send(message); return new PassThroughPromiseToListenableFutureAdapter(promise); } @Override public void onReadInactivity(Runnable runnable, long inactivityDuration) { - this.reactorTcpConnection.on().readIdle(inactivityDuration, runnable); + this.channel.on().readIdle(inactivityDuration, runnable); } @Override public void onWriteInactivity(Runnable runnable, long inactivityDuration) { - this.reactorTcpConnection.on().writeIdle(inactivityDuration, runnable); + this.channel.on().writeIdle(inactivityDuration, runnable); } @Override public void close() { - this.reactorTcpConnection.close(); + this.channel.close(); } } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java index 7fa15c1610..2fa50726ae 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java @@ -19,18 +19,13 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; -import org.springframework.context.ApplicationEvent; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.messaging.Message; import org.springframework.messaging.StubMessageChannel; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.simp.SimpMessageType; -import org.springframework.messaging.simp.broker.BrokerAvailabilityEvent; import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.tcp.ReconnectStrategy; import org.springframework.messaging.tcp.TcpConnection; @@ -139,7 +134,7 @@ public class StompBrokerRelayMessageHandlerTests { } - private static ListenableFutureTask getFuture() { + private static ListenableFutureTask getVoidFuture() { ListenableFutureTask futureTask = new ListenableFutureTask<>(new Callable() { @Override public Void call() throws Exception { @@ -150,6 +145,17 @@ public class StompBrokerRelayMessageHandlerTests { return futureTask; } + private static ListenableFutureTask getBooleanFuture() { + ListenableFutureTask futureTask = new ListenableFutureTask<>(new Callable() { + @Override + public Boolean call() throws Exception { + return null; + } + }); + futureTask.run(); + return futureTask; + } + private static class StubTcpOperations implements TcpOperations { @@ -159,18 +165,18 @@ public class StompBrokerRelayMessageHandlerTests { @Override public ListenableFuture connect(TcpConnectionHandler connectionHandler) { connectionHandler.afterConnected(this.connection); - return getFuture(); + return getVoidFuture(); } @Override public ListenableFuture connect(TcpConnectionHandler connectionHandler, ReconnectStrategy reconnectStrategy) { connectionHandler.afterConnected(this.connection); - return getFuture(); + return getVoidFuture(); } @Override - public ListenableFuture shutdown() { - return getFuture(); + public ListenableFuture shutdown() { + return getBooleanFuture(); } } @@ -183,7 +189,7 @@ public class StompBrokerRelayMessageHandlerTests { @Override public ListenableFuture send(Message message) { this.messages.add(message); - return getFuture(); + return getVoidFuture(); } @Override