diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClient.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClient.java index ae8f1c3f0b..958556196a 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClient.java @@ -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 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(host, port, new StompReactorNettyCodec()); } /** @@ -60,6 +53,7 @@ public class ReactorNettyTcpStompClient extends StompClientSupport { * @param tcpClient the client to use */ public ReactorNettyTcpStompClient(TcpOperations 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 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> { - - private final StompEncoder encoder; - - public EncodingConsumer(StompEncoder encoder) { - this.encoder = encoder; - } - - @Override - public void accept(ByteBuf byteBuf, Message message) { - byteBuf.writeBytes(this.encoder.encode(message)); - } - } - - private static final class DecodingFunction implements Function>> { - - private final StompDecoder decoder; - - public DecodingFunction(StompDecoder decoder) { - this.decoder = decoder; - } - - @Override - public List> apply(ByteBuf buffer) { - return this.decoder.decode(buffer.nioBuffer()); - } - } } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java index b8b9c98a98..9f0dd38076 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java @@ -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 codec = new StompReactorNettyCodec(decoder); + this.tcpClient = new ReactorNettyTcpClient<>(this.relayHost, this.relayPort, codec); } if (logger.isInfoEnabled()) { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompReactorNettyCodec.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompReactorNettyCodec.java new file mode 100644 index 0000000000..8b36d4bce4 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompReactorNettyCodec.java @@ -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 { + + 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))); + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyCodec.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyCodec.java new file mode 100644 index 0000000000..3254882a94 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyCodec.java @@ -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

{ + + private final Function>> decoder; + + private final BiConsumer> encoder; + + + public ReactorNettyCodec(Function>> decoder, + BiConsumer> encoder) { + + Assert.notNull(decoder, "'decoder' is required"); + Assert.notNull(encoder, "'encoder' is required"); + this.decoder = decoder; + this.encoder = encoder; + } + + public Function>> getDecoder() { + return this.decoder; + } + + public BiConsumer> getEncoder() { + return this.encoder; + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java index 75df13921b..a53c02e045 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java @@ -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

implements TcpOperations

{ private final TcpClient tcpClient; - private final MessageHandlerConfiguration

configuration; + private final ReactorNettyCodec

codec; + + private final Scheduler scheduler = Schedulers.newParallel("ReactorNettyTcpClient"); private final ChannelGroup group; @@ -76,14 +77,14 @@ public class ReactorNettyTcpClient

implements TcpOperations

{ * 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 {@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

configuration) { - this(opts -> opts.connect(host, port), configuration); + public ReactorNettyTcpClient(String host, int port, ReactorNettyCodec

codec) { + this(opts -> opts.connect(host, port), codec); } /** @@ -93,15 +94,15 @@ public class ReactorNettyTcpClient

implements TcpOperations

{ * configuration. * * @param tcpOptions callback for configuring shared {@link ClientOptions} - * @param configuration the client configuration + * @param codec for encoding and decoding messages */ public ReactorNettyTcpClient(Consumer tcpOptions, - MessageHandlerConfiguration

configuration) { + ReactorNettyCodec

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

implements TcpOperations

{ } Mono 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

implements TcpOperations

{ MonoProcessor 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

implements TcpOperations

{ this.stopping = true; - Mono completion = ChannelFutureMono.from(this.group.close()); - - if (this.configuration.scheduler != null) { - completion = completion.doAfterTerminate((x, e) -> configuration.scheduler.shutdown()); - } + Mono completion = ChannelFutureMono.from(this.group.close()) + .doAfterTerminate((x, e) -> this.scheduler.shutdown()); return new MonoToListenableFutureAdapter<>(completion); } - /** - * A configuration holder - */ - public static final class MessageHandlerConfiguration

{ - - private final Function>> decoder; - - private final BiConsumer> encoder; - - private final int backlog; - - private final Scheduler scheduler; - - - public MessageHandlerConfiguration( - Function>> decoder, - BiConsumer> 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

implements BiFunction> { private final TcpConnectionHandler

connectionHandler; - private final MessageHandlerConfiguration

configuration; + private final ReactorNettyCodec

codec; + + private final Scheduler scheduler; - MessageHandler(TcpConnectionHandler

handler, MessageHandlerConfiguration

config) { + MessageHandler(TcpConnectionHandler

handler, ReactorNettyCodec

codec, + Scheduler scheduler) { + this.connectionHandler = handler; - this.configuration = config; + this.codec = codec; + this.scheduler = scheduler; } @Override public Publisher apply(NettyInbound in, NettyOutbound out) { - Flux>> inbound = in.receive().map(configuration.decoder); + Flux>> inbound = in.receive().map(this.codec.getDecoder()); DirectProcessor closeProcessor = DirectProcessor.create(); - TcpConnection

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

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