diff --git a/build.gradle b/build.gradle index 8b5239dea1..b97c681d0b 100644 --- a/build.gradle +++ b/build.gradle @@ -117,9 +117,9 @@ configure(allprojects) { project -> } repositories { - mavenLocal() maven { url "https://repo.spring.io/libs-release" } maven { url "https://repo.spring.io/milestone" } + maven { url "https://repo.spring.io/snapshot" } // for reactor 2.0.1 snapshot } dependencies { @@ -487,6 +487,7 @@ project("spring-messaging") { compile(project(":spring-beans")) compile(project(":spring-core")) compile(project(":spring-context")) + optional("io.projectreactor:reactor-core:${reactorVersion}") optional("io.projectreactor:reactor-net:${reactorVersion}") { exclude group: "io.netty", module: "netty-all" } @@ -497,7 +498,6 @@ project("spring-messaging") { optional("org.eclipse.jetty.websocket:websocket-client:${jettyVersion}") optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}") testCompile(project(":spring-test")) - testCompile('org.slf4j:slf4j-log4j12:1.7.10') testCompile("javax.inject:javax.inject-tck:1") testCompile("javax.servlet:javax.servlet-api:3.1.0") testCompile("javax.validation:validation-api:1.0.0.GA") diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2StompCodec.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2StompCodec.java index 4ee2074da0..4f211d6fc1 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2StompCodec.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2StompCodec.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java index 59abaa32a4..23220225e1 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java @@ -18,16 +18,17 @@ package org.springframework.messaging.simp.stomp; import org.springframework.messaging.Message; import org.springframework.messaging.tcp.TcpOperations; import org.springframework.messaging.tcp.reactor.Reactor2TcpClient; +import org.springframework.messaging.tcp.reactor.Reactor2TcpClient.TcpClientSpecFactory; import org.springframework.util.concurrent.ListenableFuture; import reactor.Environment; import reactor.core.config.ConfigurationReader; import reactor.core.config.DispatcherConfiguration; import reactor.core.config.DispatcherType; import reactor.core.config.ReactorConfiguration; -import reactor.fn.Function; -import reactor.io.net.Spec; +import reactor.io.net.Spec.TcpClientSpec; import java.util.Arrays; +import java.util.List; import java.util.Properties; /** @@ -52,40 +53,18 @@ public class Reactor2TcpStompClient extends StompClientSupport { /** * Create an instance with the given host and port. - * * @param host the host * @param port the port */ public Reactor2TcpStompClient(final String host, final int port) { - this.tcpClient = new Reactor2TcpClient(createNettyTcpClientFactory(host, port)); - } - - private Function, Message>, - Spec.TcpClientSpec, Message>> createNettyTcpClientFactory( - final String host, final int port - ) { - - final Environment environment = new Environment(new StompClientDispatcherConfigReader()).assignErrorJournal(); - - return new Function, Message>, - Spec.TcpClientSpec, Message>>() { - - @Override - public Spec.TcpClientSpec, Message> apply(Spec.TcpClientSpec, - Message> spec) { - - return spec - .codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder())) - .env(environment) - .dispatcher(environment.getCachedDispatchers("StompClient").get()) - .connect(host, port); - } - }; + ConfigurationReader reader = new StompClientDispatcherConfigReader(); + Environment environment = new Environment(reader).assignErrorJournal(); + StompTcpClientSpecFactory factory = new StompTcpClientSpecFactory(environment, host, port); + this.tcpClient = new Reactor2TcpClient(factory); } /** * Create an instance with a pre-configured TCP client. - * * @param tcpClient the client to use */ public Reactor2TcpStompClient(TcpOperations tcpClient) { @@ -136,8 +115,35 @@ public class Reactor2TcpStompClient extends StompClientSupport { String dispatcherName = "StompClient"; DispatcherType dispatcherType = DispatcherType.DISPATCHER_GROUP; DispatcherConfiguration config = new DispatcherConfiguration(dispatcherName, dispatcherType, 128, 0); - return new ReactorConfiguration(Arrays.asList(config), dispatcherName, new Properties - ()); + List configList = Arrays.asList(config); + return new ReactorConfiguration(configList, dispatcherName, new Properties()); + } + } + + private static class StompTcpClientSpecFactory + implements TcpClientSpecFactory, Message> { + + private final Environment environment; + + private final String host; + + private final int port; + + public StompTcpClientSpecFactory(Environment environment, String host, int port) { + this.environment = environment; + this.host = host; + this.port = port; + } + + @Override + public TcpClientSpec, Message> apply( + TcpClientSpec, Message> tcpClientSpec) { + + return tcpClientSpec + .codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder())) + .env(this.environment) + .dispatcher(this.environment.getCachedDispatchers("StompClient").get()) + .connect(this.host, this.port); } } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java index e70b0974cd..935128ae76 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java @@ -54,7 +54,8 @@ abstract class AbstractPromiseToListenableFutureAdapter implements Listena public void accept(S result) { try { registry.success(adapt(result)); - } catch (Throwable t) { + } + catch (Throwable t) { registry.failure(t); } } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java index d880cdd71d..3028307f31 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -16,12 +16,11 @@ package org.springframework.messaging.tcp.reactor; - import reactor.rx.Promise; /** - * A Promise-to-ListenableFutureAdapter where the source and the target from the Promise and - * the ListenableFuture respectively are of the same type. + * A Promise-to-ListenableFutureAdapter where the source and the target from + * the Promise and the ListenableFuture respectively are of the same type. * * @author Rossen Stoyanchev * @since 4.0 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 d71d5f1e1d..5780392c5d 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,13 +16,17 @@ package org.springframework.messaging.tcp.reactor; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; + import io.netty.channel.nio.NioEventLoopGroup; -import org.springframework.messaging.Message; -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.concurrent.ListenableFuture; +import reactor.Environment; +import reactor.core.config.ConfigurationReader; +import reactor.core.config.DispatcherConfiguration; +import reactor.core.config.ReactorConfiguration; import reactor.core.support.NamedDaemonThreadFactory; import reactor.fn.BiFunction; import reactor.fn.Consumer; @@ -34,7 +38,7 @@ import reactor.io.codec.Codec; import reactor.io.net.ChannelStream; import reactor.io.net.NetStreams; import reactor.io.net.Reconnect; -import reactor.io.net.Spec; +import reactor.io.net.Spec.TcpClientSpec; import reactor.io.net.impl.netty.NettyClientSocketOptions; import reactor.io.net.impl.netty.tcp.NettyTcpClient; import reactor.io.net.tcp.TcpClient; @@ -43,16 +47,20 @@ import reactor.rx.Stream; import reactor.rx.Streams; import reactor.rx.action.Signal; -import java.net.InetSocketAddress; -import java.util.ArrayList; -import java.util.List; +import org.springframework.messaging.Message; +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.concurrent.ListenableFuture; /** * An implementation of {@link org.springframework.messaging.tcp.TcpOperations} * based on the TCP client support of the Reactor project. - *

- * This client will wrap N number of clients for N {@link #connect} calls (one client by connection). + * + *

This implementation wraps N (Reactor) clients for N {@link #connect} calls, + * i.e. a separate (Reactor) client instance for each connection. * * @author Rossen Stoyanchev * @author Stephane Maldini @@ -63,172 +71,136 @@ public class Reactor2TcpClient

implements TcpOperations

{ @SuppressWarnings("rawtypes") public static final Class REACTOR_TCP_CLIENT_TYPE = NettyTcpClient.class; - private final Function, Message

>, - Spec.TcpClientSpec, Message

>> tcpClientSpec; + private final TcpClientSpecFactory, Message

> tcpClientSpecFactory; - private final List, Message

>> activeClients = + private final List, Message

>> tcpClients = new ArrayList, Message

>>(); + /** - * A constructor that creates a {@link reactor.io.net.Spec.TcpClientSpec} factory with - * a default {@link reactor.core.dispatch.SynchronousDispatcher} as a result of which - * network I/O is handled in Netty threads. Number of Netty threads can be tweaked with - * the {@code reactor.tcp.ioThreadCount} System property. - *

- * The network I/O threads will be shared amongst the active clients. - *

- *

+ * A constructor that creates a {@link TcpClientSpec TcpClientSpec} factory + * with a default {@link reactor.core.dispatch.SynchronousDispatcher}, i.e. + * relying on Netty threads. The number of Netty threads can be tweaked with + * the {@code reactor.tcp.ioThreadCount} System property. The network I/O + * threads will be shared amongst the active clients. *

Also see the constructor accepting a ready Reactor - * {@link reactor.io.net.Spec.TcpClientSpec} {@link Function} factory. - * + * {@link TcpClientSpec} {@link Function} factory. * @param host the host to connect to * @param port the port to connect to * @param codec the codec to use for encoding and decoding the TCP stream */ public Reactor2TcpClient(final String host, final int port, final Codec, Message

> codec) { - //FIXME Should it be exposed in Spring ? - int ioThreadCount; - try { - ioThreadCount = Integer.parseInt(System.getProperty("reactor.tcp.ioThreadCount")); - } catch (Exception i) { - ioThreadCount = -1; - } - if (ioThreadCount <= 0l) { - ioThreadCount = Runtime.getRuntime().availableProcessors(); - } + final NioEventLoopGroup eventLoopGroup = initEventLoopGroup(); - final NioEventLoopGroup eventLoopGroup = - new NioEventLoopGroup(ioThreadCount, new NamedDaemonThreadFactory("reactor-tcp-io")); - - this.tcpClientSpec = new Function, Message

>, - Spec.TcpClientSpec, Message

>>() { + this.tcpClientSpecFactory = new TcpClientSpecFactory, Message

>() { @Override - public Spec.TcpClientSpec, Message

> apply(Spec.TcpClientSpec, Message

> - messageMessageTcpClientSpec) { - return messageMessageTcpClientSpec + public TcpClientSpec, Message

> apply(TcpClientSpec, Message

> spec) { + return spec + .env(new Environment(new SynchronousDispatcherConfigReader())) .codec(codec) - //make connect dynamic or use reconnect strategy to LB onto cluster .connect(host, port) .options(new NettyClientSocketOptions().eventLoopGroup(eventLoopGroup)); } }; } + private NioEventLoopGroup initEventLoopGroup() { + int ioThreadCount; + try { + ioThreadCount = Integer.parseInt(System.getProperty("reactor.tcp.ioThreadCount")); + } + catch (Exception i) { + ioThreadCount = -1; + } + if (ioThreadCount <= 0l) { + ioThreadCount = Runtime.getRuntime().availableProcessors(); + } + + return new NioEventLoopGroup(ioThreadCount, + new NamedDaemonThreadFactory("reactor-tcp-io")); + } + /** - * A constructor with a pre-configured {@link reactor.io.net.Spec.TcpClientSpec} {@link Function} factory. - * This might be used to add SSL or specific network parameters to the generated client configuration. - *

+ * A constructor with a pre-configured {@link TcpClientSpec} {@link Function} + * factory. This might be used to add SSL or specific network parameters to + * the generated client configuration. *

NOTE: if the client is configured with a thread-creating - * dispatcher, you are responsible for cleaning them, e.g. using {@link reactor.core.Dispatcher#shutdown}. + * dispatcher, you are responsible for cleaning them, e.g. using + * {@link reactor.core.Dispatcher#shutdown}. * * @param tcpClientSpecFactory the TcpClientSpec {@link Function} to use for each client creation. */ - public Reactor2TcpClient(Function, Message

>, - Spec.TcpClientSpec, Message

>> tcpClientSpecFactory) { + public Reactor2TcpClient(TcpClientSpecFactory, Message

> tcpClientSpecFactory) { Assert.notNull(tcpClientSpecFactory, "'tcpClientClientFactory' must not be null"); - this.tcpClientSpec = tcpClientSpecFactory; + this.tcpClientSpecFactory = tcpClientSpecFactory; } @Override public ListenableFuture connect(TcpConnectionHandler

connectionHandler) { - - //create a new client - TcpClient, Message

> tcpClient = NetStreams.tcpClient(REACTOR_TCP_CLIENT_TYPE, tcpClientSpec); - - //attach connection handler + Class type = REACTOR_TCP_CLIENT_TYPE; + TcpClient, Message

> tcpClient = NetStreams.tcpClient(type, this.tcpClientSpecFactory); composeConnectionHandling(tcpClient, connectionHandler); - - //open connection Promise promise = tcpClient.open(); - - //adapt to ListenableFuture - return new AbstractPromiseToListenableFutureAdapter(promise) { - @Override - protected Void adapt(Boolean result) { - return null; - } - }; + return new BooleanToVoidAdapter(promise); } @Override - public ListenableFuture connect(final TcpConnectionHandler

connectionHandler, - final ReconnectStrategy reconnectStrategy) { - - Assert.notNull(reconnectStrategy, "ReconnectStrategy must not be null"); - - Reconnect reconnect = new Reconnect() { - @Override - public Tuple2 reconnect(InetSocketAddress address, int attempt) { - return Tuple.of(address, reconnectStrategy.getTimeToNextAttempt(attempt)); - } - }; - - //create a new client - TcpClient, Message

> tcpClient = NetStreams.tcpClient(REACTOR_TCP_CLIENT_TYPE, tcpClientSpec); - - //attach connection handler - composeConnectionHandling(tcpClient, connectionHandler); - - //open connection - Stream stream = tcpClient.open(reconnect); - - //adapt to ListenableFuture with the next available connection - Promise promise = stream.next().map( - new Function() { - @Override - public Void apply(Boolean ch) { - return null; - } - }); - - return new PassThroughPromiseToListenableFutureAdapter(promise); + public ListenableFuture connect(TcpConnectionHandler

handler, ReconnectStrategy strategy) { + Assert.notNull(strategy, "ReconnectStrategy must not be null"); + Class type = REACTOR_TCP_CLIENT_TYPE; + TcpClient, Message

> tcpClient = NetStreams.tcpClient(type, this.tcpClientSpecFactory); + composeConnectionHandling(tcpClient, handler); + Stream stream = tcpClient.open(new ReactorRectonnectAdapter(strategy)); + return new BooleanToVoidAdapter(stream.next()); } private void composeConnectionHandling(final TcpClient, Message

> tcpClient, - final TcpConnectionHandler

connectionHandler) { + final TcpConnectionHandler

connectionHandler) { - synchronized (activeClients){ - activeClients.add(tcpClient); + synchronized (this.tcpClients){ + this.tcpClients.add(tcpClient); } tcpClient - .finallyDo(new Consumer,Message

>>>() { + .finallyDo(new Consumer, Message

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

>> signal) { - synchronized (activeClients) { - activeClients.remove(tcpClient); + public void accept(Signal, Message

>> signal) { + synchronized (tcpClients) { + tcpClients.remove(tcpClient); } - if(signal.isOnError()) { + if (signal.isOnError()) { connectionHandler.afterConnectFailure(signal.getThrowable()); } } }) - .log("reactor.client") .consume(new Consumer, Message

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

> connection) { connection - .log("reactor.connection") .finallyDo(new Consumer>>() { + @Override public void accept(Signal> signal) { if (signal.isOnError()) { connectionHandler.handleFailure(signal.getThrowable()); - } else if (signal.isOnComplete()) { + } + else if (signal.isOnComplete()) { connectionHandler.afterConnectionClosed(); } } }) .consume(new Consumer>() { + @Override public void accept(Message

message) { connectionHandler.handleMessage(message); } }); - connectionHandler.afterConnected(new Reactor2TcpConnection

(connection)); } }); @@ -236,12 +208,11 @@ public class Reactor2TcpClient

implements TcpOperations

{ @Override public ListenableFuture shutdown() { + final List, Message

>> clients; - synchronized (activeClients){ - clients = new ArrayList, Message

>>(activeClients); - //should be cleared individually in tcpClient#finallyDo() - //activeClients.clear(); + synchronized (this.tcpClients) { + clients = new ArrayList, Message

>>(this.tcpClients); } Promise promise = Streams.from(clients) @@ -259,11 +230,45 @@ public class Reactor2TcpClient

implements TcpOperations

{ }) .next(); - return new AbstractPromiseToListenableFutureAdapter(promise) { - @Override - protected Boolean adapt(Boolean result) { - return result; - } - }; + return new PassThroughPromiseToListenableFutureAdapter(promise); } + + + public interface TcpClientSpecFactory extends Function, TcpClientSpec> { + } + + private static class SynchronousDispatcherConfigReader implements ConfigurationReader { + + @Override + public ReactorConfiguration read() { + return new ReactorConfiguration(Arrays.asList(), "sync", new Properties()); + } + } + + private static class ReactorRectonnectAdapter implements Reconnect { + + private final ReconnectStrategy strategy; + + public ReactorRectonnectAdapter(ReconnectStrategy strategy) { + this.strategy = strategy; + } + + @Override + public Tuple2 reconnect(InetSocketAddress address, int attempt) { + return Tuple.of(address, strategy.getTimeToNextAttempt(attempt)); + } + } + + private static class BooleanToVoidAdapter extends AbstractPromiseToListenableFutureAdapter { + + public BooleanToVoidAdapter(Promise promise) { + super(promise); + } + + @Override + protected Void adapt(Boolean result) { + return null; + } + } + } \ No newline at end of file diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpConnection.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpConnection.java index 2f9df2b432..a08533b702 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpConnection.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpConnection.java @@ -16,64 +16,57 @@ package org.springframework.messaging.tcp.reactor; -import org.reactivestreams.Publisher; +import reactor.fn.Functions; +import reactor.io.net.ChannelStream; +import reactor.rx.Promises; +import reactor.rx.broadcast.Broadcaster; + import org.springframework.messaging.Message; import org.springframework.messaging.tcp.TcpConnection; import org.springframework.util.concurrent.ListenableFuture; -import reactor.fn.Consumer; -import reactor.fn.Function; -import reactor.fn.Functions; -import reactor.io.net.ChannelStream; -import reactor.rx.Promise; -import reactor.rx.Promises; -import reactor.rx.Stream; -import reactor.rx.broadcast.Broadcaster; - -import java.lang.reflect.Constructor; /** - * An implementation of {@link org.springframework.messaging.tcp.TcpConnection} - * based on the TCP client support of the Reactor project. + * An implementation of {@link org.springframework.messaging.tcp.TcpConnection + * TcpConnection} based on the TCP client support of the Reactor project. + * + * @param

the payload type of messages read or written to the TCP stream. * * @author Rossen Stoyanchev - * @since 4.0 - * @param

the payload type of Spring Message's read from and written to - * the TCP stream + * @since 4.2 */ public class Reactor2TcpConnection

implements TcpConnection

{ - private final ChannelStream, Message

> channel; - private final Broadcaster> sink; + private final ChannelStream, Message

> channelStream; + + private final Broadcaster> sink; - public Reactor2TcpConnection(ChannelStream, Message

> connection) { - this.channel = connection; + public Reactor2TcpConnection(ChannelStream, Message

> channelStream) { + this.channelStream = channelStream; this.sink = Broadcaster.create(); - - channel.sink(sink); + this.channelStream.sink(this.sink); } @Override public ListenableFuture send(Message

message) { - sink.onNext(message); - //FIXME need to align Reactor with Reactive IPC to have publish/confirm receipt + this.sink.onNext(message); return new PassThroughPromiseToListenableFutureAdapter(Promises.success(null)); } @Override public void onReadInactivity(Runnable runnable, long inactivityDuration) { - this.channel.on().readIdle(inactivityDuration, Functions.consumer(runnable)); + this.channelStream.on().readIdle(inactivityDuration, Functions.consumer(runnable)); } @Override public void onWriteInactivity(Runnable runnable, long inactivityDuration) { - this.channel.on().writeIdle(inactivityDuration, Functions.consumer(runnable)); + this.channelStream.on().writeIdle(inactivityDuration, Functions.consumer(runnable)); } @Override public void close() { - sink.onComplete(); + this.sink.onComplete(); } }