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

Upgrade to Reactor Netty 0.8

Browse Source 
Issue: SPR-16387
This commit is contained in:
Violeta Georgieva
2018-05-17 21:47:18 +03:00
committed by Rossen Stoyanchev
parent 61ffbe5554
commit ffbc75ae47
20 changed files with 144 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java
View File

@@ -39,14 +39,13 @@ import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.ipc.netty.FutureMono;
import reactor.ipc.netty.NettyContext;
import reactor.ipc.netty.NettyInbound;
import reactor.ipc.netty.NettyOutbound;
import reactor.ipc.netty.options.ClientOptions;
import reactor.ipc.netty.resources.LoopResources;
import reactor.ipc.netty.resources.PoolResources;
import reactor.ipc.netty.tcp.TcpClient;
import reactor.netty.Connection;
import reactor.netty.FutureMono;
import reactor.netty.NettyInbound;
import reactor.netty.NettyOutbound;
import reactor.netty.resources.ConnectionProvider;
import reactor.netty.resources.LoopResources;
import reactor.netty.tcp.TcpClient;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -83,7 +82,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
private LoopResources loopResources;
@Nullable
private PoolResources poolResources;
private ConnectionProvider poolResources;
private final Scheduler scheduler = Schedulers.newParallel("tcp-client-scheduler");
@@ -98,57 +97,18 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
* @see org.springframework.messaging.simp.stomp.StompReactorNettyCodec
*/
public ReactorNettyTcpClient(String host, int port, ReactorNettyCodec<P> codec) {
this(builder -> builder.host(host).port(port), codec);
}
/**
* Constructor with a {@link ClientOptions.Builder} that can be used to
* customize Reactor Netty client options.
*
* <p><strong>Note: </strong> this constructor manages the lifecycle of the
* {@link TcpClient} and its underlying resources. Please do not customize
* any of the following options:
* {@link ClientOptions.Builder#channelGroup(ChannelGroup) ChannelGroup},
* {@link ClientOptions.Builder#loopResources(LoopResources) LoopResources}, and
* {@link ClientOptions.Builder#poolResources(PoolResources) PoolResources}.
* You may set the {@link ClientOptions.Builder#disablePool() disablePool}
* option if you simply want to turn off pooling.
*
* <p>For full control over the initialization and lifecycle of the TcpClient,
* see {@link #ReactorNettyTcpClient(TcpClient, ReactorNettyCodec)}.
*
* @param optionsConsumer consumer to customize client options
* @param codec the code to use
* @see org.springframework.messaging.simp.stomp.StompReactorNettyCodec
*/
public ReactorNettyTcpClient(Consumer<ClientOptions.Builder<?>> optionsConsumer,
ReactorNettyCodec<P> codec) {
Assert.notNull(optionsConsumer, "Consumer<ClientOptions.Builder<?> is required");
Assert.notNull(host, "host is required");
Assert.notNull(port, "port is required");
Assert.notNull(codec, "ReactorNettyCodec is required");
this.channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
Consumer<ClientOptions.Builder<?>> builtInConsumer = builder -> {
Assert.isTrue(!builder.isLoopAvailable() && !builder.isPoolAvailable(),
"The provided ClientOptions.Builder contains LoopResources and/or PoolResources. " +
"Please, use the constructor that accepts a TcpClient instance " +
"for full control over initialization and lifecycle.");
builder.channelGroup(this.channelGroup);
builder.preferNative(false);
this.loopResources = LoopResources.create("tcp-client-loop");
builder.loopResources(this.loopResources);
if (!builder.isPoolDisabled()) {
this.poolResources = PoolResources.elastic("tcp-client-pool");
builder.poolResources(this.poolResources);
}
};
this.tcpClient = TcpClient.create(optionsConsumer.andThen(builtInConsumer));
this.loopResources = LoopResources.create("tcp-client-loop");
this.poolResources = ConnectionProvider.elastic("tcp-client-pool");
this.tcpClient = TcpClient.create(poolResources)
.host(host)
.port(port)
.runOn(loopResources, false)
.doOnConnected(c -> channelGroup.add(c.channel()));
this.codec = codec;
}
@@ -181,7 +141,8 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
}
Mono<Void> connectMono = this.tcpClient
.newHandler(new ReactorNettyHandler(handler))
.handle(new ReactorNettyHandler(handler))
.connect()
.doOnError(handler::afterConnectFailure)
.then();
@@ -201,11 +162,12 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
MonoProcessor<Void> connectMono = MonoProcessor.create();
this.tcpClient
.newHandler(new ReactorNettyHandler(handler))
.handle(new ReactorNettyHandler(handler))
.connect()
.doOnNext(updateConnectMono(connectMono))
.doOnError(updateConnectMono(connectMono))
.doOnError(handler::afterConnectFailure) // report all connect failures to the handler
.flatMap(NettyContext::onClose) // post-connect issues
.flatMap(Connection::onDispose) // post-connect issues
.retryWhen(reconnectFunction(strategy))
.repeatWhen(reconnectFunction(strategy))
.subscribe();
@@ -302,14 +264,16 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
@Override
@SuppressWarnings("unchecked")
public Publisher<Void> apply(NettyInbound inbound, NettyOutbound outbound) {
if (logger.isDebugEnabled()) {
logger.debug("Connected to " + inbound.remoteAddress());
}
inbound.withConnection(c -> {
if (logger.isDebugEnabled()) {
logger.debug("Connected to " + c.address());
}
});
DirectProcessor<Void> completion = DirectProcessor.create();
TcpConnection<P> connection = new ReactorNettyTcpConnection<>(inbound, outbound, codec, completion);
scheduler.schedule(() -> connectionHandler.afterConnected(connection));
inbound.context().addHandler(new StompMessageDecoder<>(codec));
inbound.withConnection(c -> c.addHandler(new StompMessageDecoder<>(codec)));
inbound.receiveObject()
.cast(Message.class)

17
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpConnection.java
View File

@@ -17,12 +17,10 @@
package org.springframework.messaging.tcp.reactor;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelPipeline;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Mono;
import reactor.ipc.netty.NettyInbound;
import reactor.ipc.netty.NettyOutbound;
import reactor.ipc.netty.NettyPipeline;
import reactor.netty.NettyInbound;
import reactor.netty.NettyOutbound;
import org.springframework.messaging.Message;
import org.springframework.messaging.tcp.TcpConnection;
@@ -66,20 +64,13 @@ public class ReactorNettyTcpConnection<P> implements TcpConnection<P> {
@Override
@SuppressWarnings("deprecation")
public void onReadInactivity(Runnable runnable, long inactivityDuration) {
// TODO: workaround for https://github.com/reactor/reactor-netty/issues/22
ChannelPipeline pipeline = this.inbound.context().channel().pipeline();
String name = NettyPipeline.OnChannelReadIdle;
if (pipeline.context(name) != null) {
pipeline.remove(name);
}
this.inbound.onReadIdle(inactivityDuration, runnable);
this.inbound.withConnection(c -> c.onReadIdle(inactivityDuration, runnable));
}
@Override
@SuppressWarnings("deprecation")
public void onWriteInactivity(Runnable runnable, long inactivityDuration) {
this.outbound.onWriteIdle(inactivityDuration, runnable);
this.inbound.withConnection(c -> c.onWriteIdle(inactivityDuration, runnable));
}
@Override