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

Simplify ReactorNettyTcpClient input

Browse Source 
Create a ReactorNettyCodec to hold the decoding and encoding function
and consumer along with a package-private sub-class that delegates to
StompDecoder and StompEncoder.

Issue: SPR-14531
This commit is contained in:
Rossen Stoyanchev
2016-11-30 21:11:19 -05:00
parent 85c93f5d67
commit b874692452
5 changed files with 142 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClient.java
View File

@@ -16,16 +16,9 @@
package org.springframework.messaging.simp.stomp;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Function;
import io.netty.buffer.ByteBuf;
import reactor.core.scheduler.Schedulers;
import org.springframework.messaging.Message;
import org.springframework.messaging.tcp.TcpOperations;
import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
/**
@@ -38,6 +31,7 @@ public class ReactorNettyTcpStompClient extends StompClientSupport {
private final TcpOperations<byte[]> tcpClient;
/**
* Create an instance with host "127.0.0.1" and port 61613.
*/
@@ -45,14 +39,13 @@ public class ReactorNettyTcpStompClient extends StompClientSupport {
this("127.0.0.1", 61613);
}
/**
* Create an instance with the given host and port.
* @param host the host
* @param port the port
*/
public ReactorNettyTcpStompClient(final String host, final int port) {
this.tcpClient = create(host, port, new StompDecoder());
public ReactorNettyTcpStompClient(String host, int port) {
this.tcpClient = new ReactorNettyTcpClient<byte[]>(host, port, new StompReactorNettyCodec());
}
/**
@@ -60,6 +53,7 @@ public class ReactorNettyTcpStompClient extends StompClientSupport {
* @param tcpClient the client to use
*/
public ReactorNettyTcpStompClient(TcpOperations<byte[]> tcpClient) {
Assert.notNull(tcpClient, "'tcpClient' is required");
this.tcpClient = tcpClient;
}
@@ -94,49 +88,4 @@ public class ReactorNettyTcpStompClient extends StompClientSupport {
this.tcpClient.shutdown();
}
/**
* Create a new {@link ReactorNettyTcpClient} with Stomp specific configuration for
* encoding, decoding and hand-off.
*
* @param host target host
* @param port target port
* @param decoder {@link StompDecoder} to use
* @return a new {@link TcpOperations}
*/
protected static TcpOperations<byte[]> create(String host, int port, StompDecoder decoder) {
return new ReactorNettyTcpClient<>(host, port,
new ReactorNettyTcpClient.MessageHandlerConfiguration<>(
new DecodingFunction(decoder),
new EncodingConsumer(new StompEncoder()),
128,
Schedulers.newParallel("StompClient")));
}
private static final class EncodingConsumer implements BiConsumer<ByteBuf, Message<byte[]>> {
private final StompEncoder encoder;
public EncodingConsumer(StompEncoder encoder) {
this.encoder = encoder;
}
@Override
public void accept(ByteBuf byteBuf, Message<byte[]> message) {
byteBuf.writeBytes(this.encoder.encode(message));
}
}
private static final class DecodingFunction implements Function<ByteBuf, List<Message<byte[]>>> {
private final StompDecoder decoder;
public DecodingFunction(StompDecoder decoder) {
this.decoder = decoder;
}
@Override
public List<Message<byte[]>> apply(ByteBuf buffer) {
return this.decoder.decode(buffer.nioBuffer());
}
}
}

5
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java
View File

@@ -40,6 +40,7 @@ import org.springframework.messaging.tcp.FixedIntervalReconnectStrategy;
import org.springframework.messaging.tcp.TcpConnection;
import org.springframework.messaging.tcp.TcpConnectionHandler;
import org.springframework.messaging.tcp.TcpOperations;
import org.springframework.messaging.tcp.reactor.ReactorNettyCodec;
import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
@@ -387,8 +388,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
if (this.tcpClient == null) {
StompDecoder decoder = new StompDecoder();
decoder.setHeaderInitializer(getHeaderInitializer());
this.tcpClient = ReactorNettyTcpStompClient.create(this.relayHost, this.relayPort, decoder);
ReactorNettyCodec<byte[]> codec = new StompReactorNettyCodec(decoder);
this.tcpClient = new ReactorNettyTcpClient<>(this.relayHost, this.relayPort, codec);
}
if (logger.isInfoEnabled()) {

42
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompReactorNettyCodec.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2002-2016 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.simp.stomp;
import org.springframework.messaging.tcp.reactor.ReactorNettyCodec;
/**
* {@code ReactorNettyCodec} that delegates to {@link StompDecoder} and
* {@link StompEncoder}.
*
* @author Rossen Stoyanchev
* @since 5.0
*/
class StompReactorNettyCodec extends ReactorNettyCodec<byte[]> {
public StompReactorNettyCodec() {
this(new StompDecoder(), new StompEncoder());
}
public StompReactorNettyCodec(StompDecoder decoder) {
this(decoder, new StompEncoder());
}
public StompReactorNettyCodec(StompDecoder decoder, StompEncoder encoder) {
super(byteBuf -> decoder.decode(byteBuf.nioBuffer()),
(byteBuf, message) -> byteBuf.writeBytes(encoder.encode(message)));
}
}

58
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyCodec.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2002-2016 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.tcp.reactor;
import java.util.Collection;
import java.util.function.BiConsumer;
import java.util.function.Function;
import io.netty.buffer.ByteBuf;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* Simple holder for a decoding {@link Function} and an encoding
* {@link BiConsumer} to use with Reactor Netty.
*
* @author Rossen Stoyanchev
* @since 5.0
*/
public class ReactorNettyCodec<P> {
private final Function<? super ByteBuf, ? extends Collection<Message<P>>> decoder;
private final BiConsumer<? super ByteBuf, ? super Message<P>> encoder;
public ReactorNettyCodec(Function<? super ByteBuf, ? extends Collection<Message<P>>> decoder,
BiConsumer<? super ByteBuf, ? super Message<P>> encoder) {
Assert.notNull(decoder, "'decoder' is required");
Assert.notNull(encoder, "'encoder' is required");
this.decoder = decoder;
this.encoder = encoder;
}
public Function<? super ByteBuf, ? extends Collection<Message<P>>> getDecoder() {
return this.decoder;
}
public BiConsumer<? super ByteBuf, ? super Message<P>> getEncoder() {
return this.encoder;
}
}

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

@@ -17,12 +17,10 @@
package org.springframework.messaging.tcp.reactor;
import java.util.Collection;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import io.netty.buffer.ByteBuf;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.ImmediateEventExecutor;
@@ -32,6 +30,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.ipc.netty.ChannelFutureMono;
import reactor.ipc.netty.NettyContext;
import reactor.ipc.netty.NettyInbound;
@@ -63,7 +62,9 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
private final TcpClient tcpClient;
private final MessageHandlerConfiguration<P> configuration;
private final ReactorNettyCodec<P> codec;
private final Scheduler scheduler = Schedulers.newParallel("ReactorNettyTcpClient");
private final ChannelGroup group;
@@ -76,14 +77,14 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
* the {@code reactor.tcp.ioThreadCount} System property. The network I/O
* threads will be shared amongst the active clients.
* <p>Also see the constructor accepting a {@link Consumer} of
* {@link ClientOptions} for advanced tuning.
* {@link ClientOptions} for additional options.
*
* @param host the host to connect to
* @param port the port to connect to
* @param configuration the client configuration
* @param codec for encoding and decoding messages
*/
public ReactorNettyTcpClient(String host, int port, MessageHandlerConfiguration<P> configuration) {
this(opts -> opts.connect(host, port), configuration);
public ReactorNettyTcpClient(String host, int port, ReactorNettyCodec<P> codec) {
this(opts -> opts.connect(host, port), codec);
}
/**
@@ -93,15 +94,15 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
* configuration.
*
* @param tcpOptions callback for configuring shared {@link ClientOptions}
* @param configuration the client configuration
* @param codec for encoding and decoding messages
*/
public ReactorNettyTcpClient(Consumer<? super ClientOptions> tcpOptions,
MessageHandlerConfiguration<P> configuration) {
ReactorNettyCodec<P> codec) {
Assert.notNull(configuration, "'configuration' is required");
Assert.notNull(codec, "'codec' is required");
this.group = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
this.tcpClient = TcpClient.create(opts -> tcpOptions.accept(opts.channelGroup(group)));
this.configuration = configuration;
this.codec = codec;
}
@@ -116,7 +117,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
}
Mono<Void> connectMono = this.tcpClient
.newHandler(new MessageHandler<>(handler, this.configuration))
.newHandler(new MessageHandler<>(handler, this.codec, this.scheduler))
.doOnError(handler::afterConnectFailure)
.then();
@@ -136,7 +137,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
MonoProcessor<Void> connectMono = MonoProcessor.create();
this.tcpClient.newHandler(new MessageHandler<>(handler, this.configuration))
this.tcpClient.newHandler(new MessageHandler<>(handler, this.codec, this.scheduler))
.doOnNext(item -> {
if (!connectMono.isTerminated()) {
connectMono.onComplete();
@@ -163,75 +164,48 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
this.stopping = true;
Mono<Void> completion = ChannelFutureMono.from(this.group.close());
if (this.configuration.scheduler != null) {
completion = completion.doAfterTerminate((x, e) -> configuration.scheduler.shutdown());
}
Mono<Void> completion = ChannelFutureMono.from(this.group.close())
.doAfterTerminate((x, e) -> this.scheduler.shutdown());
return new MonoToListenableFutureAdapter<>(completion);
}
/**
* A configuration holder
*/
public static final class MessageHandlerConfiguration<P> {
private final Function<? super ByteBuf, ? extends Collection<Message<P>>> decoder;
private final BiConsumer<? super ByteBuf, ? super Message<P>> encoder;
private final int backlog;
private final Scheduler scheduler;
public MessageHandlerConfiguration(
Function<? super ByteBuf, ? extends Collection<Message<P>>> decoder,
BiConsumer<? super ByteBuf, ? super Message<P>> encoder,
int backlog, Scheduler scheduler) {
this.decoder = decoder;
this.encoder = encoder;
this.backlog = backlog > 0 ? backlog : QueueSupplier.SMALL_BUFFER_SIZE;
this.scheduler = scheduler;
}
}
private static final class MessageHandler<P>
implements BiFunction<NettyInbound, NettyOutbound, Publisher<Void>> {
private final TcpConnectionHandler<P> connectionHandler;
private final MessageHandlerConfiguration<P> configuration;
private final ReactorNettyCodec<P> codec;
private final Scheduler scheduler;
MessageHandler(TcpConnectionHandler<P> handler, MessageHandlerConfiguration<P> config) {
MessageHandler(TcpConnectionHandler<P> handler, ReactorNettyCodec<P> codec,
Scheduler scheduler) {
this.connectionHandler = handler;
this.configuration = config;
this.codec = codec;
this.scheduler = scheduler;
}
@Override
public Publisher<Void> apply(NettyInbound in, NettyOutbound out) {
Flux<Collection<Message<P>>> inbound = in.receive().map(configuration.decoder);
Flux<Collection<Message<P>>> inbound = in.receive().map(this.codec.getDecoder());
DirectProcessor<Void> closeProcessor = DirectProcessor.create();
TcpConnection<P> tcpConnection =
new ReactorNettyTcpConnection<>(in, out, configuration.encoder, closeProcessor);
if (configuration.scheduler != null) {
configuration.scheduler.schedule(() -> connectionHandler.afterConnected(tcpConnection));
inbound = inbound.publishOn(configuration.scheduler, configuration.backlog);
}
else {
connectionHandler.afterConnected(tcpConnection);
}
TcpConnection<P> tcpConnection =
new ReactorNettyTcpConnection<>(in, out, this.codec.getEncoder(), closeProcessor);
this.scheduler.schedule(() -> connectionHandler.afterConnected(tcpConnection));
inbound = inbound.publishOn(this.scheduler, QueueSupplier.SMALL_BUFFER_SIZE);
inbound.flatMapIterable(Function.identity())
.subscribe(connectionHandler::handleMessage,
connectionHandler::handleFailure,
connectionHandler::afterConnectionClosed);
.subscribe(
connectionHandler::handleMessage,
connectionHandler::handleFailure,
connectionHandler::afterConnectionClosed);
return closeProcessor;
}